pyncette package

Module contents

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

Bases: object

Pyncette application.

create() → AsyncIterator[PyncetteContext][source]

Creates the execution context.

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

Decorator for marking the coroutine as a dynamic task

fixture(name: Optional[str] = None) → Callable[[pyncette.model.FixtureFunc], pyncette.model.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

task(**kwargs) → Callable[[pyncette.model.TaskFunc], pyncette.model.TaskFunc][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.

class pyncette.FailureMode[source]

Bases: enum.Enum

What should happen when a task fails.

NONE = 0
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

initialize(root_context: pyncette.model.Context) → None[source]
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: pyncette.task.Task, instance_name: Optional[str] = None) → None[source]

Removes the concrete instance of a dynamic task