pyncette package

Module contents

class pyncette.Pyncette(repository_factory: pyncette.repository.RepositoryFactory = <function sqlite_repository>, executor_cls: type = <class 'pyncette.executor.DefaultExecutor'>, poll_interval: datetime.timedelta = datetime.timedelta(seconds=1), **kwargs)[source]

Bases: object

Pyncette application.

create(context_items: dict[str, Any] | None = None) → AsyncIterator[PyncetteContext][source]

Creates the execution context.

dynamic_task(**kwargs) → Callable[[pyncette.model.TaskFunc], pyncette.task.Task][source]

Decorator for marking the coroutine as a dynamic task

fixture(name: str | None = None) → Decorator[FixtureFunc][source]

Decorator for marking the generator as a fixture

main() → None[source]

Convenience entrypoint for console apps, which sets up logging and signal handling.

middleware(func: pyncette.model.MiddlewareFunc) → pyncette.model.MiddlewareFunc[source]

Decorator for marking the function as a middleware

partitioned_task(**kwargs) → Callable[[pyncette.model.TaskFunc], pyncette.task.PartitionedTask][source]

Decorator for marking the coroutine as a partitioned dynamic task

task(**kwargs) → Callable[[pyncette.model.TaskFunc], pyncette.task.Task][source]

Decorator for marking the coroutine as a task

use_fixture(fixture_name: str, func: pyncette.model.FixtureFunc) → None[source]
use_middleware(func: pyncette.model.MiddlewareFunc) → None[source]
class pyncette.ExecutionMode[source]

Bases: enum.Enum

The execution mode for a Pyncette task.

AT_LEAST_ONCE = 0
AT_MOST_ONCE = 1
class pyncette.FailureMode[source]

Bases: enum.Enum

What should happen when a task fails.

COMMIT = 2
NONE = 0
UNLOCK = 1
class pyncette.Context[source]

Bases: object

Task execution context. This class can have dynamic attributes.

class pyncette.PyncetteContext(app: pyncette.pyncette.Pyncette, repository: pyncette.repository.Repository, executor: pyncette.executor.DefaultExecutor)[source]

Bases: object

Execution context of a Pyncette app

add_to_context(name: str, value: Any) → None[source]

Adds a value with a given key to task context

initialize(root_context: pyncette.model.Context) → None[source]
last_tick
run() → None[source]

Runs the Pyncette’s main event loop.

schedule_task(task: pyncette.task.Task, instance_name: str, **kwargs) → pyncette.task.Task[source]

Schedules a concrete instance of a dynamic task

shutdown() → None[source]

Initiates graceful shutdown, terminating the main loop, but allowing all executing tasks to finish.

unschedule_task(task: Task, instance_name: str | None = None) → None[source]

Removes the concrete instance of a dynamic task