API Reference

This page is automatically generated from the Python source code docstrings.

pyncette

Context

Task execution context. This class can have dynamic attributes.

app_context instance-attribute

app_context: PyncetteContext

task instance-attribute

task: Task

scheduled_at instance-attribute

scheduled_at: datetime

heartbeat instance-attribute

heartbeat: Heartbeater

args instance-attribute

args: dict[str, Any]

ExecutionMode

Bases: Enum

The execution mode for a Pyncette task.

AT_LEAST_ONCE class-attribute instance-attribute

AT_LEAST_ONCE = 0

AT_MOST_ONCE class-attribute instance-attribute

AT_MOST_ONCE = 1

FailureMode

Bases: Enum

What should happen when a task fails.

NONE class-attribute instance-attribute

NONE = 0

UNLOCK class-attribute instance-attribute

UNLOCK = 1

COMMIT class-attribute instance-attribute

COMMIT = 2

Pyncette

Pyncette(
    repository_factory: RepositoryFactory = sqlite_repository,
    executor_cls: type = DefaultExecutor,
    poll_interval: timedelta = timedelta(seconds=1),
    **kwargs: Any,
)

Pyncette application.

task

task(**kwargs: Any) -> Callable[[TaskFunc], Task]

Decorator for marking the coroutine as a task

dynamic_task

dynamic_task(**kwargs: Any) -> Callable[[TaskFunc], Task]

Decorator for marking the coroutine as a dynamic task

partitioned_task

partitioned_task(**kwargs: Any) -> Callable[[TaskFunc], PartitionedTask]

Decorator for marking the coroutine as a partitioned dynamic task

use_fixture

use_fixture(fixture_name: str, func: FixtureFunc) -> None

use_middleware

use_middleware(func: MiddlewareFunc) -> None

fixture

fixture(name: str | None = None) -> Decorator[FixtureFunc]

Decorator for marking the generator as a fixture

middleware

middleware(func: MiddlewareFunc) -> MiddlewareFunc

Decorator for marking the function as a middleware

create async

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

Creates the execution context.

main

main() -> None

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

PyncetteContext

PyncetteContext(
    app: Pyncette, repository: Repository, executor: DefaultExecutor
)

Execution context of a Pyncette app

last_tick property

last_tick: datetime | None

initialize

initialize(root_context: Context) -> None

schedule_task async

schedule_task(
    task: Task, instance_name: str, force: bool = False, **kwargs: Any
) -> Task

Schedules a concrete instance of a dynamic task

Args: task: The dynamic task to schedule instance_name: Name for this concrete instance force: If False (default), fails if task is locked or preserves sooner schedule. If True, overwrites everything including locks and schedule. **kwargs: Additional arguments for the task instance (schedule, interval, etc.)

Returns: The concrete task instance

Raises: PyncetteException: If force=False and task is currently locked

unschedule_task async

unschedule_task(task: Task, instance_name: str | None = None) -> None

Removes the concrete instance of a dynamic task

get_task async

get_task(task: Task, instance_name: str | None = None) -> TaskState | None

Get the state of a task instance

Args: task: For static tasks, the task itself. For dynamic tasks, either: - The parent task (template) with instance_name provided - An already-instantiated concrete dynamic task (instance_name should be None) instance_name: The instance name for dynamic tasks when passing the parent task. Must be None for static tasks or pre-instantiated dynamic tasks.

Returns: TaskState if found, None otherwise

Raises: ValueError: If instance_name is provided with a non-dynamic task or an already-instantiated task

list_tasks async

list_tasks(
    task: Task,
    limit: int | None = None,
    continuation_token: ContinuationToken | None = None,
) -> ListTasksResponse

List all instances of a dynamic task with pagination

Args: task: The dynamic task template limit: Maximum number of tasks to return (backend may return less) continuation_token: Token from previous response to get next page

Returns: ListTasksResponse with tasks and optional continuation token

Raises: ValueError: If task is not dynamic

Note: Results are not guaranteed to be in any particular order. Backends may order results to provide stable pagination.

add_to_context

add_to_context(name: str, value: Any) -> None

Adds a value with a given key to task context

run async

run() -> None

Runs the Pyncette's main event loop.

shutdown

shutdown() -> None

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

options: show_submodules: true