pyncette.pyncette module¶
-
class
pyncette.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
-
-
class
pyncette.pyncette.
PyncetteContext
(app: pyncette.pyncette.Pyncette, repository: pyncette.repository.Repository, executor: pyncette.executor.DefaultExecutor)[source]¶ Bases:
object
Execution context of a Pyncette app
-
last_tick
¶
-
schedule_task
(task: pyncette.task.Task, instance_name: str, **kwargs) → pyncette.task.Task[source]¶ Schedules a concrete instance of a dynamic task
-