Skip to content
QuadKit Python Framework

API Reference

Protocols

Section titled “Protocols”

source
Protocol for tracking and broadcasting task progress.

Implementations must be safe for concurrent use — multiple coroutines may call update on the same task_id simultaneously, and multiple consumers may subscribe to the same task.

The subscribe method returns a live AsyncIterator that yields a new ProgressSnapshot every time update, complete, or fail is called for the given task. The iterator terminates automatically when the task reaches a terminal state (COMPLETE or FAILED).

Example

async def export_data(tracker: ProgressTrackerProtocol) -> None:
    records = await fetch_all()
    total = len(records)
    for i, record in enumerate(records, start=1):
        await process(record)
        await tracker.update("export-1", i, total, f"Row {i}/{total}")
    await tracker.complete("export-1", "Export finished")
update
async def update(
    task_id: str,
    current: int,
    total: int,
    message: str = ''
) -> None
source

Record incremental progress for a task.

Parameters
ParameterTypeDescription
task_idstrUnique identifier for the task.
currentintUnits completed so far.
totalintTotal units to process (0 = unknown).
messagestrOptional human-readable status message.
complete
async def complete(
    task_id: str,
    result: str = '',
    metadata: Mapping[str, object] | None = None
) -> None
source

Mark a task as successfully completed.

Closes all active subscriptions for task_id after broadcasting the terminal ProgressSnapshot.

Parameters
ParameterTypeDescription
task_idstrUnique identifier for the task.
resultstrOptional human-readable completion message.
metadataMapping[str, object] | NoneOptional JSON-safe terminal UI hints.
fail
async def fail(
    task_id: str,
    error: str,
    metadata: Mapping[str, object] | None = None
) -> None
source

Mark a task as failed.

Closes all active subscriptions for task_id after broadcasting the terminal ProgressSnapshot.

Parameters
ParameterTypeDescription
task_idstrUnique identifier for the task.
errorstrDescription of the failure.
metadataMapping[str, object] | NoneOptional JSON-safe terminal UI hints.
get
async def get(task_id: str) -> ProgressSnapshot | None
source

Return the current progress state for a task.

Parameters
ParameterTypeDescription
task_idstrUnique identifier for the task.
Returns
TypeDescription
ProgressSnapshot | NoneThe most recent ProgressSnapshot, or `None` if the task is not known to this tracker.
subscribe
def subscribe(task_id: str) -> AsyncIterator[ProgressSnapshot]
source

Subscribe to live progress updates for a task.

Returns an async iterator that yields a new ProgressSnapshot each time the task’s state changes. The iterator stops automatically when the task reaches a terminal state (COMPLETE or FAILED).

If the task is already in a terminal state when subscribe is called, the iterator yields the final snapshot once and then stops.

Parameters
ParameterTypeDescription
task_idstrUnique identifier for the task to observe.
Returns
TypeDescription
AsyncIterator[ProgressSnapshot]An AsyncIterator of ProgressSnapshot objects.

Example

async for snap in tracker.subscribe("job-42"):
    print(f"{snap.percent:.1f}% — {snap.message}")

source
Protocol for task executor implementations.

Executors process tasks from a queue by dispatching them to registered handlers.

Example

class TaskExecutorProtocol:
    async def execute(self, task: Task) -> Any:
        handler = self._handlers.get(task.name)
        if handler:
            return await handler(task.payload)
        raise UnknownTaskError(task.name)
execute
async def execute(task: Any) -> Any
source

Execute a task using the registered handler.

Parameters
ParameterTypeDescription
taskAnyTask to execute.
Returns
TypeDescription
AnyTask execution result.
Raises
ExceptionDescription
UnknownTaskErrorIf no handler registered for task.
register_handler
def register_handler(
    task_name: str,
    handler: Any
) -> None
source

Register a handler for a task type.

Parameters
ParameterTypeDescription
task_namestrName of the task type.
handlerAnyAsync callable to handle the task.
get_handler
def get_handler(task_name: str) -> Any | None
source

Get the handler for a task type.

Parameters
ParameterTypeDescription
task_namestrName of the task type.
Returns
TypeDescription
Any | NoneHandler if registered, None otherwise.

source
Protocol for task queue implementations.

This defines the contract for background task queues (Redis-based, PostgreSQL-based, in-memory, etc.).

Example

class RedisTaskQueue:
    async def enqueue(self, task: Task) -> str:
        task_id = str(uuid4())
        await self._redis.rpush("tasks", task.to_json())
        return task_id


    async def dequeue(self) -> Task | None:
        data = await self._redis.lpop("tasks")
        return Task.from_json(data) if data else None
enqueue
async def enqueue(task: Any) -> Result[str, TaskQueueError]
source

Add a task to the queue.

Parameters
ParameterTypeDescription
taskAnyTask instance to enqueue.
Returns
TypeDescription
Result[str, TaskQueueError]Ok(task_id) on success; Err(TaskQueueError) if the queue reports a recoverable failure (e.g. full queue, invalid payload). Only infrastructure failures (lost connection) are raised as exceptions.
dequeue
async def dequeue() -> Any | None
source

Remove and return the next task.

Returns
TypeDescription
Any | NoneNext Task or None if queue is empty.
get_task_count
async def get_task_count() -> int
source

Get the number of tasks in the queue.

Returns
TypeDescription
intNumber of pending tasks.
clear
async def clear() -> None
source

Clear all tasks from the queue.

get_job
async def get_job(job_id: str) -> Any | None
source

Get a job by ID.

Parameters
ParameterTypeDescription
job_idstrUnique job identifier.
Returns
TypeDescription
Any | NoneJob data or None if not found.
delete_job
async def delete_job(job_id: str) -> bool
source

Delete a job from the queue.

Parameters
ParameterTypeDescription
job_idstrUnique job identifier.
Returns
TypeDescription
boolTrue if the job was deleted.
count
async def count(
    queue_name: str,
    status: Any | None = None
) -> int
source

Count queued jobs.

Parameters
ParameterTypeDescription
queue_namestrQueue to inspect.
statusAny | NoneOptional status filter.
Returns
TypeDescription
intNumber of matching jobs.
ack
async def ack(task_id: str) -> None
source

Acknowledge successful processing of a dequeued task.

Signals that the task has been processed successfully and can be permanently discarded from any in-flight tracking.

Parameters
ParameterTypeDescription
task_idstrID of the task to acknowledge.
nack
async def nack(
    task_id: str,
    requeue: bool = True
) -> None
source

Negative-acknowledge a dequeued task.

Signals that the task could not be processed. The task is optionally requeued for another processing attempt.

Parameters
ParameterTypeDescription
task_idstrID of the task to negative-acknowledge.
requeueboolIf True, return the task to the queue for retry. If False, discard the task permanently.
close
async def close() -> None
source

Close the queue connection and cleanup resources.

Classes

Section titled “Classes”

source
Tracks and shuts down background asyncio tasks.

Implements TaskManagerProtocol. Inject as a singleton from any DI provider. Call shutdown() during application stop to cancel all in-flight tasks.

Example

manager = BackgroundTaskManager()
task = manager.track(some_coroutine())
# … later …
await manager.shutdown(timeout=30.0)
manager = BackgroundTaskManager()
task = manager.track(some_coroutine())
# … later …
await manager.shutdown(timeout=30.0)
__init__
def __init__() -> None
source
track
def track(coro: Awaitable[T]) -> asyncio.Task[T]
source

Schedule coro as an asyncio task and track the handle.

The task is removed from the tracked set automatically via a done_callback so completed tasks don’t accumulate.

Parameters
ParameterTypeDescription
coroAwaitable[T]Any awaitable to wrap in a task.
Returns
TypeDescription
asyncio.Task[T]The created asyncio.Task.
track_named
def track_named(
    name: str,
    coro: Awaitable[T]
) -> asyncio.Task[T]
source

Like track but attach a local asyncio task name.

The name is available through the returned task’s introspection API, but is deliberately omitted from operational logs.

Parameters
ParameterTypeDescription
namestrLabel surfaced in observability/log output.
coroAwaitable[T]Any awaitable to wrap in a task.
Returns
TypeDescription
asyncio.Task[T]The created asyncio.Task.
pending_count
property pending_count() -> int
source

Number of tasks that have not yet completed.

shutdown
async def shutdown(timeout: float = 30.0) -> None
source

Cancel all pending tracked tasks and await them.

After cancellation each task is given up to timeout seconds collectively to finish (honour CancelledError and finalise any finally blocks). A count-only warning is emitted if any tasks remain; task names are never logged.

Parameters
ParameterTypeDescription
timeoutfloatSeconds to wait for all tasks to finish after cancellation. Default 30 seconds.

source
Branch based on a condition evaluated against the input.

Example

workflow = BranchStep( condition=lambda data: data[“amount”] > 1000, if_true=TaskStep(“manual_review”, manual_review), if_false=TaskStep(“auto_approve”, auto_approve), )

__init__
def __init__(
    condition: Callable[[Any], bool],
    if_true: TaskStep,
    if_false: TaskStep
) -> None
source
execute
async def execute(input_data: Any = None) -> Result[WorkflowResult, WorkflowError]
source

Evaluate the condition and execute either the true or false branch.

source
Distributed progress store backed by the platform ``CacheBackendProtocol``.

Stores serialised ProgressInfo objects in the configured cache backend so that progress is visible to all worker processes. TTL-based expiry is delegated to the backend.

Parameters
ParameterTypeDescription
cachePlatform cache backend (Redis, Memcached, …).
ttlSeconds before a progress entry expires. Defaults to 3600 (1 hour).
__init__
def __init__(
    cache: CacheBackendProtocol,
    *,
    ttl: int = 3600
) -> None
source
save
async def save(info: ProgressInfo) -> None
source

Persist a ProgressInfo entry to the cache backend.

Parameters
ParameterTypeDescription
infoProgressInfoProgress state to save.
get
async def get(job_id: str) -> ProgressInfo | None
source

Return the ProgressInfo for the given job ID, or None if not found.

Parameters
ParameterTypeDescription
job_idstrUnique identifier for the job.
delete
async def delete(job_id: str) -> None
source

Remove the progress entry for the given job ID.

Parameters
ParameterTypeDescription
job_idstrUnique identifier for the job.
list_active
async def list_active() -> list[ProgressInfo]
source

Return all active (incomplete) progress entries.

Note This implementation scans the key prefix via get, which may be slow on large datasets. Consider using a Redis-specific implementation with SCAN for production at scale.

Returns
TypeDescription
list[ProgressInfo]List of incomplete ProgressInfo.

source
Distributed result store backed by the platform ``CacheBackendProtocol``.

Stores serialised JobResult objects in the configured cache backend so that results are visible to all worker processes. TTL-based expiry is delegated to the backend.

Parameters
ParameterTypeDescription
cachePlatform cache backend (Redis, Memcached, …).
ttlSeconds before a result entry expires. Defaults to 3600 (1 hour).
poll_intervalSeconds between polls in wait. Defaults to 0.5.
__init__
def __init__(
    cache: CacheBackendProtocol,
    *,
    ttl: int = 3600,
    poll_interval: float = 0.5
) -> None
source
store
async def store(
    job_id: str,
    result: JobResult
) -> None
source

Persist a job result in the cache backend with TTL.

get
async def get(job_id: str) -> JobResult | None
source

Return the stored result, or None if absent or expired.

delete
async def delete(job_id: str) -> bool
source

Remove the result from the cache and return True if it existed.

get_completed
async def get_completed(limit: int | None = None) -> list[JobResult]
source

Return recently completed results, newest first.

Parameters
ParameterTypeDescription
limitint | NoneMaximum number of results to return; `None` for all.
Returns
TypeDescription
list[JobResult]Successful results stored through this backend, newest first.
get_failed
async def get_failed(limit: int | None = None) -> list[JobResult]
source

Return recently failed results, newest first.

Parameters
ParameterTypeDescription
limitint | NoneMaximum number of results to return; `None` for all.
Returns
TypeDescription
list[JobResult]Failed results stored through this backend, newest first.
wait
async def wait(
    job_id: str,
    *,
    timeout: float = 30.0,
    poll_interval: float | None = None
) -> JobResult | None
source

Poll the cache until the result appears or the timeout elapses.

cleanup_expired
async def cleanup_expired() -> int
source

No-op: the cache backend handles TTL expiry automatically.

source
Wrapper for cron expression parsing.

Provides a unified interface for cron expression handling with optional croniter dependency.

Note This class requires the croniter optional dependency. See quadkit.tasks.scheduling.cron for installation instructions. Instantiation raises :exc:ImportError when croniter is absent.

Example

cron = CronExpression("0 9 * * 1-5")  # Weekdays at 9am
next_run = cron.get_next()
__init__
def __init__(expression: str)
source

Initialize cron expression

Parameters
ParameterTypeDescription
expressionstrCron expression string
Raises
ExceptionDescription
ImportErrorIf croniter is not installed
ValueErrorIf expression is invalid
get_next
def get_next(base_time: float | None = None) -> float
source

Get next execution time

Parameters
ParameterTypeDescription
base_timefloat | NoneBase time (defaults to current time)
Returns
TypeDescription
floatUnix timestamp of next execution
get_prev
def get_prev(base_time: float | None = None) -> float
source

Get previous execution time

Parameters
ParameterTypeDescription
base_timefloat | NoneBase time (defaults to current time)
Returns
TypeDescription
floatUnix timestamp of previous execution

source
Manages failed jobs for inspection, retry, and cleanup.
__init__
def __init__(
    *,
    max_size: int = 10000,
    retention_hours: float = 168
) -> None
source
add
def add(
    job: JobProtocol,
    error: str,
    traceback: str | None = None,
    **metadata: Any
) -> None
source

Add a failed job to the DLQ.

Parameters
ParameterTypeDescription
jobJobProtocolThe failed job.
errorstrError message.
tracebackstr | NoneFull traceback string. **metadata: Additional metadata.
get
def get(job_id: str) -> FailureRecord | None
source

Get a specific failure record.

list_failed
def list_failed(
    *,
    limit: int = 50,
    job_name: str | None = None
) -> list[FailureRecord]
source

List failed jobs, optionally filtered by job name.

Parameters
ParameterTypeDescription
limitintMaximum records to return.
job_namestr | NoneFilter by job type name.
Returns
TypeDescription
list[FailureRecord]List of failure records (newest first).
retry
def retry(job_id: str) -> JobProtocol | None
source

Remove a job from DLQ and prepare it for retry.

Returns the job with status reset to PENDING, or None if not found.

retry_all
def retry_all() -> list[JobProtocol]
source

Retry all failed jobs. Returns list of jobs to re-enqueue.

remove
def remove(job_id: str) -> bool
source

Permanently remove a failure from the DLQ.

purge
def purge(
    *,
    older_than_hours: float | None = None
) -> int
source

Purge old failures.

Parameters
ParameterTypeDescription
older_than_hoursfloat | NoneRemove records older than this. Defaults to retention_hours.
Returns
TypeDescription
intNumber of records purged.
clear
def clear() -> int
source

Clear all DLQ entries. Returns count cleared.

size
property size() -> int
source

Return the number of currently tracked failure records.

get_stats
def get_stats() -> dict[str, Any]
source

Get DLQ statistics.

source
In-memory lock for preventing duplicate task execution within a single process.

Note: This is NOT a distributed lock. For true distributed locking across multiple processes/servers, use a Redis-based or database-based lock implementation.

Example

lock_manager = LockManager()
async with lock_manager.acquire("user:123:sync", timeout=60):
    await sync_user_data(user_id=123)
__init__
def __init__(
    key: str,
    timeout: float,
    locks_dict: dict[str, tuple[float, asyncio.Lock]],
    locks_lock: asyncio.Lock
)
source

Initialize in-memory lock.

Parameters
ParameterTypeDescription
keystrUnique lock key
timeoutfloatLock expiry in seconds; the key is considered expired once this many seconds have passed since acquisition and is purged on the next acquire attempt
locks_dictdict[str, tuple[float, asyncio.Lock]]Shared locks dictionary (from LockManager)
locks_lockasyncio.LockShared lock for accessing locks_dict (from LockManager)
acquire
async def acquire() -> bool
source

Try to acquire the lock without waiting.

Expired lock entries are purged before the ownership check, so an expired key can be re-acquired immediately.

Returns
TypeDescription
boolTrue if the lock was acquired, False if the key is already held.
try_acquire
async def try_acquire() -> bool
source

Try to acquire lock without waiting.

Returns
TypeDescription
boolTrue if acquired, False if already held
acquire_wait
async def acquire_wait(timeout: float | None = None) -> bool
source

Block until the lock is acquired or timeout seconds elapse.

Polls acquire with capped exponential backoff between attempts, so the caller never busy-waits. Each poll purges expired lock entries, so a timed-out holder loses the key to a waiting caller.

Parameters
ParameterTypeDescription
timeoutfloat | NoneMaximum seconds to wait for the lock. When `None`, wait until the lock is acquired or the calling task is cancelled.
Returns
TypeDescription
boolTrue when the lock was acquired, False if *timeout* elapsed without acquiring it.

Example

lock = lock_manager.acquire("user:123:sync", timeout=60)
if await lock.acquire_wait(timeout=30):
    try:
        await sync_user_data(user_id=123)
    finally:
        await lock.release()
release
async def release() -> None
source

Release the lock.

Only the registry entry that still belongs to this instance is removed: a stale release — after this lock’s entry expired and the key was re-acquired by a different caller — leaves the current holder’s entry untouched and only frees this instance’s own asyncio.Lock.

source
Record of a single task execution.
to_dict
def to_dict() -> dict[str, Any]
source

Serialize the execution record to a plain dictionary.

source
Record of a failed job in the DLQ.
to_dict
def to_dict() -> dict[str, Any]
source

Serialize the failure record to a plain dictionary.

from_dict
def from_dict(
    cls,
    data: dict[str, Any]
) -> FailureRecord
source

Deserialize a failure record from a plain dictionary.

Parameters
ParameterTypeDescription
datadict[str, Any]Dictionary as produced by to_dict.
Returns
TypeDescription
FailureRecordReconstructed FailureRecord.

source
System-wide throughput cap applied across all queues.

Delegates to a single TokenBucket for consistent rate enforcement regardless of which queue a task originates from.

Example

limiter = GlobalRateLimiter(rate=100, per=1.0)  # 100 tasks / sec
await limiter.acquire()
limiter = GlobalRateLimiter(rate=100, per=1.0)  # 100 tasks / sec
await limiter.acquire()
__init__
def __init__(
    rate: int,
    per: float = 1.0,
    burst: int | None = None
) -> None
source

Initialise the global rate limiter.

Parameters
ParameterTypeDescription
rateintMaximum allowed tasks per time period.
perfloatLength of the time period in seconds.
burstint | NoneMaximum burst capacity (defaults to *rate*).
acquire
async def acquire() -> None
source

Block until a global token is available.

try_acquire
async def try_acquire() -> bool
source

Non-blocking global token acquisition.

Returns
TypeDescription
bool`True if a token was obtained; False` if globally rate-limited.

source
Registry for task handlers that also satisfies the TaskExecutorProtocol contract.

Maps task names to their handler functions. Handlers can be registered and retrieved by name. The execute method dispatches a task to its registered handler, making this class usable wherever a TaskExecutorProtocol is expected.

Extends Registry for unified introspection and lifecycle hooks.

Example

handler = registry.get("send_email")
```python
__init__
def __init__() -> None
source

Initialize handler registry.

register
def register(
    name: str,
    handler: TaskHandler | None = None
) -> Callable[Ellipsis, Any]
source

Register a task handler.

Can be used as a decorator or called directly.

Parameters
ParameterTypeDescription
namestrTask name.
handlerTaskHandler | NoneHandler function (if not using as decorator).
Returns
TypeDescription
Callable[Ellipsis, Any]Handler function or decorator.

Example

# As decorator
@registry.register("task_name")
async def my_handler(...):
    pass


# Direct call
registry.register("task_name", my_handler)
register_handler
def register_handler(
    task_name: str,
    handler: Any
) -> None
source

Register a handler for a task type (TaskExecutorProtocol method).

This method satisfies the register_handler requirement defined in quadkit.contracts.infra.tasks.protocols.TaskExecutorProtocol. It is a protocol-mandated entry point that delegates directly to register so that HandlerRegistry can be used wherever a TaskExecutorProtocol is expected without any additional adapter. Do not remove — the protocol contract requires this exact signature.

Parameters
ParameterTypeDescription
task_namestrName of the task type.
handlerAnyAsync callable to handle the task.
get_handler
def get_handler(task_name: str) -> Any | None
source

Get the handler for a task type (TaskExecutorProtocol method).

This method satisfies the get_handler requirement defined in quadkit.contracts.infra.tasks.protocols.TaskExecutorProtocol. It is a protocol-mandated entry point that delegates directly to get so that HandlerRegistry can be used wherever a TaskExecutorProtocol is expected without any additional adapter. Do not remove — the protocol contract requires this exact signature.

Parameters
ParameterTypeDescription
task_namestrName of the task type.
Returns
TypeDescription
Any | NoneHandler if registered, `None` otherwise.
execute
async def execute(task: Any) -> Any
source

Execute a task using the registered handler (TaskExecutorProtocol protocol).

Resolves the handler for task.name, then invokes it with task.args and task.kwargs. Supports both sync and async handlers.

Parameters
ParameterTypeDescription
taskAnyTask object with `name, args, and kwargs` attributes.
Returns
TypeDescription
AnyResult returned by the handler.
Raises
ExceptionDescription
TaskNotFoundErrorIf no handler is registered for the task name.
remove
def remove(name: str) -> bool
source

Remove handler from registry.

Parameters
ParameterTypeDescription
namestrTask name.
Returns
TypeDescription
boolTrue if handler was removed.
list_handlers
def list_handlers() -> list[str]
source

List all registered handler names.

Returns
TypeDescription
list[str]List of task names.
to_dict
def to_dict() -> dict[str, Callable[Ellipsis, Awaitable[Any]]]
source

Get handlers as dictionary.

Returns
TypeDescription
dict[str, Callable[Ellipsis, Awaitable[Any]]]Dictionary of all handlers.

source
In-memory progress store for development/testing.
__init__
def __init__() -> None
source
save
async def save(info: ProgressInfo) -> None
source

Persist a ProgressInfo entry in the in-memory store.

get
async def get(job_id: str) -> ProgressInfo | None
source

Return the ProgressInfo for the given job ID, or None if not found.

delete
async def delete(job_id: str) -> None
source

Remove the progress entry for the given job ID if it exists.

list_active
async def list_active() -> list[ProgressInfo]
source

Return all progress entries whose current count has not yet reached total.

source
Thread-safe, in-memory ProgressTrackerProtocol implementation.

Maintains the latest ProgressSnapshot for each tracked task and fans out every state change to all active subscribers via asyncio.Queue.

Subscriptions are kept alive until the task reaches a terminal state (COMPLETE or FAILED), at which point the iterator closes automatically. Subscribers that call subscribe after the task has already finished receive the terminal snapshot once and then stop.

All public methods are safe to call concurrently from multiple coroutines.

__init__
def __init__() -> None
source
update
async def update(
    task_id: str,
    current: int,
    total: int,
    message: str = ''
) -> None
source

Record incremental progress and broadcast to all subscribers.

Parameters
ParameterTypeDescription
task_idstrUnique identifier for the task.
currentintUnits completed so far.
totalintTotal units to process (0 means unknown).
messagestrOptional human-readable status line.
complete
async def complete(
    task_id: str,
    result: str = '',
    metadata: Mapping[str, object] | None = None
) -> None
source

Mark a task as successfully completed and close all subscriptions.

Parameters
ParameterTypeDescription
task_idstrUnique identifier for the task.
resultstrOptional human-readable completion message.
metadataMapping[str, object] | NoneOptional JSON-safe terminal UI hints.
fail
async def fail(
    task_id: str,
    error: str,
    metadata: Mapping[str, object] | None = None
) -> None
source

Mark a task as failed and close all subscriptions.

Parameters
ParameterTypeDescription
task_idstrUnique identifier for the task.
errorstrFailure text; private details are reduced to a generic safe message.
metadataMapping[str, object] | NoneOptional JSON-safe terminal UI hints.
get
async def get(task_id: str) -> ProgressSnapshot | None
source

Return the current progress state for a task.

Parameters
ParameterTypeDescription
task_idstrUnique identifier for the task.
Returns
TypeDescription
ProgressSnapshot | NoneThe most recent ProgressSnapshot, or `None` if the task has not been seen by this tracker.
subscribe
def subscribe(task_id: str) -> AsyncGenerator[ProgressSnapshot, None]
source

Subscribe to live progress updates for a task.

Returns an async generator that yields one ProgressSnapshot per state change. The generator stops automatically when the task reaches a terminal state. If the task is already finished, the terminal snapshot is yielded once and the generator stops immediately.

Parameters
ParameterTypeDescription
task_idstrUnique identifier for the task to observe.
Returns
TypeDescription
AsyncGenerator[ProgressSnapshot, None]An async generator of ProgressSnapshot objects.

source
In-memory result store with TTL-based expiry.

Suitable for development and single-process deployments. For production, implement a Redis-backed store.

__init__
def __init__(
    *,
    ttl: int = 3600,
    max_size: int = 10000
) -> None
source
store
async def store(
    job_id: str,
    result: JobResult
) -> None
source

Persist a job result, evicting the oldest entry when at capacity.

get
async def get(job_id: str) -> JobResult | None
source

Return the result for the given job ID, or None if absent or expired.

delete
async def delete(job_id: str) -> bool
source

Remove the result for the given job ID and return True if it existed.

wait
async def wait(
    job_id: str,
    *,
    timeout: float = 30.0,
    poll_interval: float = 0.5
) -> JobResult | None
source

Wait for a result to appear, with timeout.

cleanup_expired
async def cleanup_expired() -> int
source

Remove all entries whose TTL has elapsed and return the count removed.

get_stats
def get_stats() -> dict[str, Any]
source

Return store statistics including size, capacity, and waiter count.

get_completed
async def get_completed(limit: int | None = None) -> list[JobResult]
source

Return the most recently completed results, newest first.

Parameters
ParameterTypeDescription
limitint | NoneMaximum number of results to return; `None` for all.
Returns
TypeDescription
list[JobResult]Successful stored results, newest first.
get_failed
async def get_failed(limit: int | None = None) -> list[JobResult]
source

Return the most recently failed results, newest first.

Parameters
ParameterTypeDescription
limitint | NoneMaximum number of results to return; `None` for all.
Returns
TypeDescription
list[JobResult]Failed stored results, newest first.

source
Represents a background job with comprehensive status tracking

Jobs are executable work units that can be queued, scheduled, tracked, and retried. They support priority ordering, dependency management, and timeout control.

Attributes: id: Unique job identifier name: Human-readable job name/type args: Positional arguments for job handler kwargs: Keyword arguments for job handler priority: Execution priority (higher = more important) max_retries: Maximum retry attempts on failure timeout: Execution timeout in seconds depends_on: List of job IDs this job depends on status: Current job status created_at: Timestamp when job was created started_at: Timestamp when job started executing completed_at: Timestamp when job finished retry_count: Current number of retry attempts last_error: Most recent error message result: JobProtocol execution result scheduled_at: Timestamp for scheduled execution cron_expression: Cron expression for recurring jobs

is_pending
property is_pending() -> bool
source

Check if job is pending execution

is_running
property is_running() -> bool
source

Check if job is currently running

is_completed
property is_completed() -> bool
source

Check if job completed successfully

is_failed
property is_failed() -> bool
source

Check if job failed

can_retry
property can_retry() -> bool
source

Check if job can be retried (hasn’t exceeded max_retries)

duration_ms
property duration_ms() -> float | None
source

Get job execution duration in milliseconds

mark_running
def mark_running() -> None
source

Mark job as running and record start time

mark_completed
def mark_completed(result: JobResult) -> None
source

Mark job as completed with result

Parameters
ParameterTypeDescription
resultJobResultJobProtocol execution result
mark_failed
def mark_failed(error: str) -> None
source

Mark job as failed with error message

Parameters
ParameterTypeDescription
errorstrError message describing the failure
mark_retrying
def mark_retrying() -> None
source

Mark job as retrying and increment retry count

mark_cancelled
def mark_cancelled() -> None
source

Mark job as cancelled

to_dict
def to_dict() -> dict[str, Any]
source

Convert job to dictionary for serialization

Returns
TypeDescription
dict[str, Any]Dictionary representation of job
from_dict
def from_dict(
    cls,
    data: dict[str, Any]
) -> JobProtocol
source

Create job from dictionary

Parameters
ParameterTypeDescription
datadict[str, Any]Dictionary representation of job
Returns
TypeDescription
JobProtocolJobProtocol instance

source
Result of a job execution

Captures the outcome of job execution including success status, return data, error information, and performance metrics.

ok
def ok(
    cls,
    data: Any = None,
    duration: float = 0.0
) -> JobResult
source

Create a successful result

Parameters
ParameterTypeDescription
dataAnyReturn value from the job
durationfloatExecution duration in seconds
Returns
TypeDescription
JobResultJobResult with success=True
fail
def fail(
    cls,
    error: str,
    retry_count: int = 0,
    duration: float = 0.0
) -> JobResult
source

Create a failure result

Parameters
ParameterTypeDescription
errorstrError message describing the failure
retry_countintNumber of retry attempts made
durationfloatExecution duration in seconds
Returns
TypeDescription
JobResultJobResult with success=False
to_dict
def to_dict() -> dict[str, Any]
source

Convert result to dictionary for serialization

from_dict
def from_dict(
    cls,
    data: dict[str, Any]
) -> JobResult
source

Create result from dictionary

source
JobProtocol execution status.

source
Template for creating scheduled jobs

JobTemplateProtocol defines a reusable job configuration that can be instantiated multiple times with different IDs.

Attributes: name: JobProtocol name/type args: Positional arguments kwargs: Keyword arguments priority: Execution priority max_retries: Maximum retry attempts timeout: Execution timeout depends_on: JobProtocol dependencies template

create_job
def create_job(job_id: str = '') -> JobProtocol
source

Create a JobProtocol instance from this template

Parameters
ParameterTypeDescription
job_idstrJobProtocol ID (generated if empty)
Returns
TypeDescription
JobProtocolNew JobProtocol instance
from_job
def from_job(
    cls,
    job: JobProtocol
) -> JobTemplateProtocol
source

Create template from existing job

Parameters
ParameterTypeDescription
jobJobProtocolJobProtocol to convert to template
Returns
TypeDescription
JobTemplateProtocolJobTemplateProtocol instance

source
Logs task start, completion, and failures.
before_execute
async def before_execute(ctx: TaskExecutionContext) -> None
source

Log task start before execution.

after_execute
async def after_execute(ctx: TaskExecutionContext) -> None
source

Log task completion status and duration after execution.

on_error
async def on_error(
    ctx: TaskExecutionContext,
    error: Exception
) -> None
source

Log the error when a task raises an exception.

source
In-memory task queue implementation

Uses Python’s heapq for priority-based ordering. Tasks are stored in memory and will be lost on process restart. Thread-safe using asyncio.Lock for concurrent access.

Characteristics:

  • Storage: In-memory heap-based priority queue
  • Concurrency: asyncio.Lock for thread safety
  • Persistence: No persistence (volatile)
  • Use Case: Development, testing, single-process applications
  • Limitations: No persistence, single-process only

Example

queue = MemoryTaskQueue()
await queue.enqueue(task)
task = await queue.dequeue()
__init__
def __init__() -> None
source

Initialize memory task queue

set_hook_registry
def set_hook_registry(hooks: HookRegistryProtocol | None) -> None
source

Attach an optional hook registry after provider boot wiring.

enqueue
async def enqueue(task: JobProtocol) -> Result[str, TaskError]
source

Add a task to the queue, returning Ok(task_id) or Err(TaskError) on failure.

Tasks are stored in a min-heap with negative priority to achieve max-heap behavior (highest priority first).

Parameters
ParameterTypeDescription
taskJobProtocolTask to enqueue
Returns
TypeDescription
Result[str, TaskError]Ok containing the enqueued task id, or Err containing a TaskError.
dequeue
async def dequeue() -> JobProtocol | None
source

Remove and return the next task from the queue

Returns the highest priority task. If multiple tasks have the same priority, returns the oldest (FIFO). Respects task delay - tasks with future available_at are skipped.

Returns
TypeDescription
JobProtocol | NoneNext task or None if queue is empty or no tasks are available
ack
async def ack(task_id: str) -> None
source

Acknowledge successful processing of a task

Removes the task from the in-flight tracking set.

Parameters
ParameterTypeDescription
task_idstrID of the task to acknowledge
nack
async def nack(
    task_id: str,
    requeue: bool = True
) -> None
source

Negative-acknowledge a task, optionally requeuing it

Removes the task from in-flight tracking. If requeue is True the task is pushed back onto the priority heap for retry.

Parameters
ParameterTypeDescription
task_idstrID of the task to negative-acknowledge
requeueboolIf True, return the task to the queue for retry
get_task_count
async def get_task_count() -> int
source

Get the number of tasks in the queue

Returns
TypeDescription
intNumber of pending tasks
clear
async def clear() -> None
source

Clear all tasks from the queue

get_job
async def get_job(job_id: str) -> JobProtocol | None
source

Get a queued or in-flight job by ID.

delete_job
async def delete_job(job_id: str) -> bool
source

Delete a queued or in-flight job by ID.

count
async def count(
    queue_name: str,
    status: Any | None = None
) -> int
source

Count queued jobs (single-queue in-memory implementation).

close
async def close() -> None
source

Close the queue connection

For memory queue, there are no external resources to clean up.

source
Collects per-task-type execution metrics.
__init__
def __init__() -> None
source
after_execute
async def after_execute(ctx: TaskExecutionContext) -> None
source

Record execution metrics for the completed task.

on_error
async def on_error(
    ctx: TaskExecutionContext,
    error: Exception
) -> None
source

Increment failure counter for the task type that raised an error.

get_stats
def get_stats() -> dict[str, Any]
source

Return aggregated execution statistics keyed by task name.

source
Configuration for a single named task queue backend.

Used in TaskConfig.backends to declare multiple task queues that the framework registers as Named DI bindings.

Parameters
ParameterTypeDescription
nameUnique backend identifier. Used as the Named() DI key.
primaryWhether this is the primary backend. Primary backends also receive the unnamed TaskQueueProtocol binding.
typeBackend type ('memory', 'redis', 'rabbitmq', 'postgres').
redis_urlRedis connection URL (when type='redis').
amqp_urlAMQP connection URL (when type='rabbitmq').
postgres_dsnPostgres DSN (when type='postgres').
queue_nameQueue name for this backend instance.

Example

backends:
  - name: primary
    primary: true
    type: redis
    redis_url: "${REDIS_URL}"
  - name: notifications
    type: rabbitmq
    amqp_url: "${AMQP_URL}"
backends:
  - name: primary
    primary: true
    type: redis
    redis_url: "${REDIS_URL}"
  - name: notifications
    type: rabbitmq
    amqp_url: "${AMQP_URL}"

Inject by name

queue: Annotated[TaskQueueProtocol, Named("notifications")]
queue: Annotated[TaskQueueProtocol, Named("notifications")]

source
Controls what happens when a scheduled worker cycle raises.

Attributes: LOG_AND_CONTINUE: Log the error, then resume the normal schedule. BACKOFF: Log the error and double the sleep before the next cycle (up to 10 * interval_seconds). STOP: Log the error and stop the worker permanently.

source
Task priority levels for queue ordering (higher number = higher priority).

The scale is 0–20 and is separate from the kernel’s ProviderPriority (0–100). Workers dequeue higher-priority jobs before lower-priority ones.

Values: LOW (0): Background tasks that can be deferred indefinitely. NORMAL (5): Default priority for routine task processing. HIGH (10): Time-sensitive tasks processed before NORMAL queue. CRITICAL (20): Highest priority; reserved for urgent system tasks.

source
Current progress state of a task.
is_complete
property is_complete() -> bool
source

Return True when the current count has reached the total.

elapsed_seconds
property elapsed_seconds() -> float
source

Return seconds elapsed since the task started, or 0 if not started.

source
Immutable point-in-time view of a task's progress.
Parameters
ParameterTypeDescription
task_idUnique identifier for the tracked task.
currentNumber of units completed so far.
totalTotal number of units to complete (0 means unknown).
statusCurrent lifecycle state.
messageHuman-readable status message.
errorError description when `status is FAILED`; empty otherwise.
metadataJSON-safe terminal UI hints copied by tracker implementations.
percent
property percent() -> float
source

Completion percentage in the range [0.0, 100.0].

Returns 0.0 when total is unknown (zero).

source
Lifecycle states for a tracked task.

Values are lowercase strings so they round-trip cleanly through JSON.

source
Abstract storage for progress state.
save
async def save(info: ProgressInfo) -> None
source

Save progress state.

get
async def get(job_id: str) -> ProgressInfo | None
source

Get progress for a job.

delete
async def delete(job_id: str) -> None
source

Remove progress entry.

list_active
async def list_active() -> list[ProgressInfo]
source

List all active (incomplete) progress entries.

source
Track and report task progress.

Injected into task handlers to enable progress reporting.

__init__
def __init__(
    job_id: str,
    store: ProgressStore
) -> None
source
update
async def update(
    current: int,
    total: int,
    message: str = '',
    **metadata: Any
) -> None
source

Update progress.

Parameters
ParameterTypeDescription
currentintCurrent progress count.
totalintTotal items to process.
messagestrHuman-readable status message. **metadata: Additional metadata.
complete
async def complete(message: str = 'Complete') -> None
source

Mark progress as complete.

fail
async def fail(message: str) -> None
source

Mark progress as failed.

info
property info() -> ProgressInfo
source

Return the current ProgressInfo snapshot.

source
Per-queue rate limiter.

Manages independent rate limits for named task queues, delegating each limit to a TokenBucket.

Example

limiter = QueueRateLimiter()
limiter.add_limit("emails", rate=5, per=1.0)   # 5 / sec
limiter.add_limit("reports", rate=1, per=60.0) # 1 / min


await limiter.acquire("emails")
limiter = QueueRateLimiter()
limiter.add_limit("emails", rate=5, per=1.0)   # 5 / sec
limiter.add_limit("reports", rate=1, per=60.0) # 1 / min


await limiter.acquire("emails")
add_limit
def add_limit(
    queue: str,
    rate: int,
    per: float = 1.0,
    burst: int | None = None
) -> None
source

Register a rate limit for a named queue.

Parameters
ParameterTypeDescription
queuestrQueue name to apply the limit to.
rateintAllowed requests per time period.
perfloatLength of the time period in seconds.
burstint | NoneMaximum burst size (defaults to *rate*).
acquire
async def acquire(queue: str) -> None
source

Block until a token is available for queue.

Parameters
ParameterTypeDescription
queuestrQueue name; a no-op for unlisted queues.
try_acquire
async def try_acquire(queue: str) -> bool
source

Attempt a non-blocking token acquisition for queue.

Parameters
ParameterTypeDescription
queuestrQueue name.
Returns
TypeDescription
bool`True if a token was obtained or the queue has no limit; False` if the queue is currently rate-limited.

source
Abstract result store backend.
store
async def store(
    job_id: str,
    result: JobResult
) -> None
source

Store a job result.

get
async def get(job_id: str) -> JobResult | None
source

Get a result by job ID. Returns None if not found or expired.

delete
async def delete(job_id: str) -> bool
source

Delete a result. Returns True if deleted.

wait
async def wait(
    job_id: str,
    *,
    timeout: float = 30.0,
    poll_interval: float = 0.5
) -> JobResult | None
source

Wait for a result with timeout.

cleanup_expired
async def cleanup_expired() -> int
source

Remove expired results. Returns count removed.

get_completed
async def get_completed(limit: int | None = None) -> list[JobResult]
source

Return recently completed job results, newest first.

Backends that can enumerate stored results should override this; the default implementation returns an empty list.

Parameters
ParameterTypeDescription
limitint | NoneMaximum number of results to return; `None` for all.
Returns
TypeDescription
list[JobResult]Completed job results, newest first.
get_failed
async def get_failed(limit: int | None = None) -> list[JobResult]
source

Return recently failed job results, newest first.

Backends that can enumerate stored results should override this; the default implementation returns an empty list.

Parameters
ParameterTypeDescription
limitint | NoneMaximum number of results to return; `None` for all.
Returns
TypeDescription
list[JobResult]Failed job results, newest first.

source
Represents a scheduled job with cron expression

Attributes: job_template: Template for creating job instances cron_expression: Cron expression string next_run: Unix timestamp of next execution enabled: Whether job is enabled for execution

calculate_next_run
def calculate_next_run() -> float
source

Calculate the next run time based on cron expression

Returns
TypeDescription
floatUnix timestamp of next execution

source
Base class for class-based tasks executed by ``TaskWorker``.

Subclasses define an async def execute(self, ...) method declaring typed DI dependencies directly as parameters

@scheduled(cron="0 3 * * *", name="calculate-affinities")
class AffinityCalculationTask(ScheduledTask):
    async def execute(
        self,
        db: DatabaseProviderProtocol,
        cache: CacheBackendProtocol,
    ) -> None:
        ...
@scheduled(cron="0 3 * * *", name="calculate-affinities")
class AffinityCalculationTask(ScheduledTask):
    async def execute(
        self,
        db: DatabaseProviderProtocol,
        cache: CacheBackendProtocol,
    ) -> None:
        ...

ScheduledTask validates execute at class creation time via __init_subclass__ rather than declaring an untyped **kwargs method signature, allowing subclasses to declare strict typed DI parameters without Liskov # type: ignore[override] pragmas.

source
Periodic worker base class.

Subclass and implement run_cycle. The base class handles the scheduling loop, error policy, graceful shutdown, and task tracking via BackgroundTaskManager.

Class-level defaults (override per subclass or via constructor)

class MyWorker(ScheduledWorker):
    interval_seconds = 60.0
    initial_delay_seconds = 0.0
    max_jitter_seconds = 0.0
    on_error_policy = OnErrorPolicy.LOG_AND_CONTINUE
class MyWorker(ScheduledWorker):
    interval_seconds = 60.0
    initial_delay_seconds = 0.0
    max_jitter_seconds = 0.0
    on_error_policy = OnErrorPolicy.LOG_AND_CONTINUE

Constructor injection is the preferred way to customise at runtime

worker = MyWorker(
    task_manager=task_manager,
    interval_seconds=30.0,
    on_error_policy=OnErrorPolicy.BACKOFF,
)
worker = MyWorker(
    task_manager=task_manager,
    interval_seconds=30.0,
    on_error_policy=OnErrorPolicy.BACKOFF,
)
__init__
def __init__(
    task_manager: BackgroundTaskManager,
    *,
    interval_seconds: float | None = None,
    initial_delay_seconds: float | None = None,
    max_jitter_seconds: float | None = None,
    on_error_policy: OnErrorPolicy | None = None
) -> None
source

Initialise the worker.

Parameters
ParameterTypeDescription
task_managerBackgroundTaskManagerShared BackgroundTaskManager that owns the task handle and participates in framework shutdown.
interval_secondsfloat | NoneOverride interval_seconds.
initial_delay_secondsfloat | NoneOverride initial_delay_seconds.
max_jitter_secondsfloat | NoneOverride max_jitter_seconds.
on_error_policyOnErrorPolicy | NoneOverride on_error_policy.
start
async def start() -> None
source

Start the worker loop as a tracked background task.

Safe to call only once. Calling start() while already running is a no-op (logs a warning).

stop
async def stop(grace_seconds: float = 2.0) -> None
source

Signal the worker to stop and wait for it to finish.

Graceful: the stop event is set so the sleeping phase wakes up early, and the current cycle (if any) is allowed to finish. If the task has not completed within grace_seconds, it is forcibly cancelled.

Parameters
ParameterTypeDescription
grace_secondsfloatSeconds to wait for a clean stop before cancelling. Default is 2 seconds.
run_cycle
async def run_cycle() -> None
source

Override to implement one unit of periodic work.

Called once per interval. Must not swallow asyncio.CancelledError.

source
Result from a single workflow step.

source
Configuration for task queue backends.

model_config: ClassVar[ConfigDict] = ConfigDict(extra=“ignore”)

source
Sequential task execution — output of each step feeds the next.

Example

chain = TaskChain([ TaskStep(“parse”, parse_data), TaskStep(“validate”, validate_data), TaskStep(“save”, save_data), ]) result = await chain.execute(raw_input)

__init__
def __init__(
    steps: list[TaskStep],
    *,
    stop_on_error: bool = True,
    state_store: WorkflowStateStore | None = None,
    workflow_id: str | None = None
) -> None
source
execute
async def execute(input_data: Any = None) -> Result[WorkflowResult, WorkflowError]
source

Execute all steps in sequence, threading output of each step into the next.

If a workflow_id and state_store were provided at construction, this method will attempt to resume from the last saved checkpoint — any step whose name appears in the persisted state’s completed steps will be skipped, and the last completed step’s output will seed the next step.

source
Fan-out then collect — parallel steps + callback with all results.

Example

chord = TaskChord( steps=[TaskStep(“a”, fetch_a), TaskStep(“b”, fetch_b)], callback=TaskStep(“merge”, merge_results), ) result = await chord.execute()

__init__
def __init__(
    steps: list[TaskStep],
    callback: TaskStep
) -> None
source
execute
async def execute(input_data: Any = None) -> Result[WorkflowResult, WorkflowError]
source

Fan out to all group steps then call the callback with their collected results.

source
Emitted when a task finishes successfully.

Attributes: task_id: Unique identifier of the completed task. worker_id: Identifier of the worker that executed the task.

source
Payload fired when a task finishes successfully.

Attributes: task_name: Name or type label of the completed task. task_id: Unique identifier of the task instance.

source
Hierarchical root configuration for Quadkit Tasks.

Attributes: name: Configuration name (default: “tasks”) enabled: Whether tasks module is enabled backend: Task queue backend configuration worker: Task worker settings scheduler: Task scheduler settings retry: Task retry configuration rate_limit: Task rate limiting timeout: Task timeout settings extra: Extra configuration

validate_for_environment
def validate_for_environment(env: Environment | None = None) -> list[ConfigIssue]
source

Validate task configuration for the given environment.

Parameters
ParameterTypeDescription
envEnvironment | NoneTarget environment; resolved from `self.env when None`.
Returns
TypeDescription
list[ConfigIssue]List of ConfigIssue instances.
from_named
def from_named(
    cls,
    entry: NamedTaskConfig
) -> TaskConfig
source

Build a single-backend TaskConfig from a NamedTaskConfig entry.

Used internally by TasksProvider to create per-backend configs.

Parameters
ParameterTypeDescription
entryNamedTaskConfigThe named backend entry to materialise.
Returns
TypeDescription
TaskConfigA TaskConfig configured for the single named backend.

source
Aggregated task observability dashboard.
__init__
def __init__(
    *,
    history_size: int = 1000,
    window_seconds: float = 300
) -> None
source
record_execution
def record_execution(
    task_name: str,
    *,
    job_id: str = '',
    duration_ms: float = 0.0,
    success: bool = True,
    error: str | None = None,
    worker_id: str | None = None
) -> None
source

Record a task execution.

get_summary
def get_summary() -> dict[str, Any]
source

Get dashboard summary with throughput and error rates.

get_per_task_stats
def get_per_task_stats() -> dict[str, dict[str, Any]]
source

Get per-task-type statistics with latency percentiles.

get_recent_executions
def get_recent_executions(
    *,
    limit: int = 50,
    task_name: str | None = None
) -> list[dict[str, Any]]
source

Get recent execution history.

get_error_breakdown
def get_error_breakdown() -> dict[str, list[dict[str, Any]]]
source

Get recent errors grouped by task type.

reset
def reset() -> None
source

Reset all counters and history.

source
Payload fired when a task is placed on the queue.

Attributes: task_name: Name or type label of the enqueued task. queue_name: Name of the queue the task was added to.

source
Context passed through the middleware pipeline.

source
Emitted when a task fails after all retry attempts are exhausted.

Attributes: task_id: Unique identifier of the failed task. error: Human-readable description of the failure reason.

source
Payload fired when a task raises an unhandled exception.

Attributes: task_name: Name or type label of the failed task. task_id: Unique identifier of the task instance. reason: Short description or exception message.

source
Parallel task execution — all steps run concurrently.

Example

group = TaskGroup([ TaskStep(“email”, send_email), TaskStep(“sms”, send_sms), ]) result = await group.execute(user_data)

__init__
def __init__(
    steps: list[TaskStep],
    *,
    max_concurrency: int | None = None,
    state_store: WorkflowStateStore | None = None,
    workflow_id: str | None = None
) -> None
source
execute
async def execute(input_data: Any = None) -> Result[WorkflowResult, WorkflowError]
source

Run all steps concurrently, respecting max_concurrency if set.

source
Health status for task processing system

Aggregates health information from queue, workers, and scheduler.

is_healthy
property is_healthy() -> bool
source

Check if system is healthy

success_rate
property success_rate() -> float
source

Calculate job success rate

to_dict
def to_dict() -> dict[str, Any]
source

Convert to dictionary for serialization

source
Abstract base for task middleware.
before_execute
async def before_execute(ctx: TaskExecutionContext) -> None
source

Called before task execution. Override to modify or log.

after_execute
async def after_execute(ctx: TaskExecutionContext) -> None
source

Called after task execution. Override to observe or log.

on_error
async def on_error(
    ctx: TaskExecutionContext,
    error: Exception
) -> None
source

Called when task execution fails.

source
Chains middleware around task execution.
__init__
def __init__() -> None
source
add
def add(middleware: TaskMiddleware) -> None
source

Append a middleware instance to the pipeline.

execute
async def execute(
    job: JobProtocol,
    handler: Any
) -> JobResult
source

Execute a job through the middleware pipeline.

Parameters
ParameterTypeDescription
jobJobProtocolThe job to execute.
handlerAnyThe async handler function.
Returns
TypeDescription
JobResultJobResult from execution.

source
Task processing provider for Quadkit Framework.

TaskProvider integrates task processing with the framework, providing dependency injection, lifecycle management, scheduling, and worker-pool orchestration.

__init__
def __init__(
    queue: TaskQueueProtocol | None = None,
    worker_count: int | None = None,
    enable_scheduler: bool | None = None,
    middleware_pipeline: TaskMiddlewarePipeline | None = None,
    config: TaskConfig | None = None,
    task_modules: list[str] | tuple[str, Ellipsis] | None = None,
    task_packages: list[str] | tuple[str, Ellipsis] | None = None
) -> None
source

Initialize the task provider.

Parameters
ParameterTypeDescription
queueTaskQueueProtocol | NoneOptional explicit task queue implementation. When `None, the queue is built from config.backend` via TaskBackendRegistry (falling back to MemoryTaskQueue when unconfigured).
worker_countint | NoneOptional explicit worker pool size (`0 disables in-process consumption for producer-only processes). Defaults to config.worker.worker_count or 1`.
enable_schedulerbool | NoneOptional explicit scheduler flag. Defaults to `config.scheduler.enabled or True`.
middleware_pipelineTaskMiddlewarePipeline | NoneOptional middleware pipeline applied to every worker in the pool.
configTaskConfig | NoneOptional typed tasks configuration. When `None, the orchestrator may late-inject the tasks` yaml section before register.
task_moduleslist[str] | tuple[str, Ellipsis] | NoneExact Python module paths to import during boot and scan for `@task / @scheduled` callables.
task_packageslist[str] | tuple[str, Ellipsis] | NonePackage roots to import recursively during boot and scan for decorated task callables.
from_config
def from_config(
    cls,
    config: TaskConfig,
    **context: Any
) -> TaskProvider
source

Create a TaskProvider from a TaskConfig.

Context kwargs may include a pre-built queue plus optional task_modules / task_packages discovery roots. If no queue is provided, one is instantiated from config.backend via TaskBackendRegistry.

source
Emitted when a task is accepted into a processing queue.

Attributes: task_id: Unique identifier of the queued task. queue_name: Name of the queue that accepted the task.

source
Configuration for rate limiting.

model_config: ClassVar[ConfigDict] = ConfigDict(extra=“ignore”)

source
Task scheduler for cron-like job scheduling

TaskScheduler manages periodic job execution using cron expressions. Jobs are checked at regular intervals and enqueued when due.

Example

scheduler = TaskScheduler()


# Schedule job
template = JobTemplateProtocol(name="cleanup", args=())
scheduler.schedule_job(template, "0 0 * * *")  # Daily at midnight


# Start scheduler
await scheduler.start_scheduler(queue.enqueue)
__init__
def __init__(
    check_interval: float = 60.0,
    store: SchedulerStore | None = None
)
source

Initialize task scheduler.

Parameters
ParameterTypeDescription
check_intervalfloatHow often to check for due jobs (seconds).
storeSchedulerStore | NoneOptional SchedulerStore for durable persistence of job definitions and next-run times. When provided, all schedule/unschedule operations are persisted and the scheduler can be restored after restart via restore_from_store.
schedule_job_sync
def schedule_job_sync(
    job_template: JobTemplateProtocol | JobProtocol,
    cron_expression: str,
    job_id: str | None = None
) -> str
source

Schedule a job with cron expression (sync, no persistence).

For the primary async version with store persistence, use schedule_job.

Parameters
ParameterTypeDescription
job_templateJobTemplateProtocol | JobProtocolJobProtocol template or JobProtocol to schedule
cron_expressionstrCron expression (e.g., "0 9 * * 1-5")
job_idstr | NoneOptional job ID, generated if not provided
Returns
TypeDescription
strJobProtocol ID for the scheduled job
schedule_job
async def schedule_job(
    job_template: JobTemplateProtocol | JobProtocol,
    cron_expression: str,
    job_id: str | None = None
) -> str
source

Schedule a job and persist to store if one is configured.

This is the primary async method. For a synchronous variant without persistence, use schedule_job_sync.

Parameters
ParameterTypeDescription
job_templateJobTemplateProtocol | JobProtocolJobProtocol template or JobProtocol to schedule.
cron_expressionstrCron expression (e.g., `"0 9 * * 1-5"`).
job_idstr | NoneOptional job ID, generated if not provided.
Returns
TypeDescription
strJobProtocol ID for the scheduled job.
restore_from_store
async def restore_from_store() -> int
source

Restore all persisted job definitions from the configured store.

Loads every job persisted by previous scheduler instances and re-registers them in scheduled_jobs. Call this once during startup before invoking start_scheduler to resume the full scheduling state.

Returns
TypeDescription
intNumber of jobs restored.
Raises
ExceptionDescription
RuntimeErrorIf no SchedulerStore was provided at construction.
unschedule_job_sync
def unschedule_job_sync(job_id: str) -> bool
source

Remove a scheduled job (sync, no store deletion).

For the primary async version with store deletion, use unschedule_job.

Parameters
ParameterTypeDescription
job_idstrJobProtocol ID to remove
Returns
TypeDescription
boolTrue if job was removed, False if not found
unschedule_job
async def unschedule_job(job_id: str) -> bool
source

Remove a scheduled job and delete it from the store.

This is the primary async method. For a synchronous variant without store deletion, use unschedule_job_sync.

Parameters
ParameterTypeDescription
job_idstrJobProtocol ID to remove.
Returns
TypeDescription
boolTrue if job was removed, False if not found.
get_scheduled_jobs
def get_scheduled_jobs() -> list[ScheduledJob]
source

Get all scheduled jobs

Returns
TypeDescription
list[ScheduledJob]List of all scheduled jobs
get_scheduled_job
def get_scheduled_job(job_id: str) -> ScheduledJob | None
source

Get a specific scheduled job

Parameters
ParameterTypeDescription
job_idstrJobProtocol ID
Returns
TypeDescription
ScheduledJob | NoneScheduledJob or None if not found
start_scheduler
async def start_scheduler(enqueue_callback: Callable[[JobProtocol], Any]) -> None
source

Start the scheduler loop

Parameters
ParameterTypeDescription
enqueue_callbackCallable[[JobProtocol], Any]Async function to enqueue jobs when they are due
stop_scheduler
async def stop_scheduler() -> None
source

Stop the scheduler gracefully

enable_job
def enable_job(job_id: str) -> bool
source

Enable a scheduled job

Parameters
ParameterTypeDescription
job_idstrJobProtocol ID
Returns
TypeDescription
boolTrue if job was enabled
disable_job
def disable_job(job_id: str) -> bool
source

Disable a scheduled job

Parameters
ParameterTypeDescription
job_idstrJobProtocol ID
Returns
TypeDescription
boolTrue if job was disabled
get_next_run_times
def get_next_run_times() -> dict[str, float]
source

Get next run times for all enabled scheduled jobs

Returns
TypeDescription
dict[str, float]Dictionary mapping job ID to next run timestamp

source
Configuration for task scheduler.

model_config: ClassVar[ConfigDict] = ConfigDict(extra=“ignore”)

source
Payload fired when a worker picks up and begins executing a task.

Attributes: task_name: Name or type label of the task being executed. task_id: Unique identifier of the task instance.

source
A single step in a workflow.
Parameters
ParameterTypeDescription
nameHuman-readable step name.
handlerAsync callable that takes input and returns output.
on_errorOptional error handler.
timeoutOptional timeout in seconds.

source
Configuration for task timeout behavior.

source
Individual task worker for job execution

TaskWorker processes jobs from a queue using registered handlers. Each worker runs independently and tracks its own statistics.

Example

worker = TaskWorker("worker-1", queue, handlers)
await worker.start()
# ... worker processes jobs ...
await worker.stop()
__init__
def __init__(
    worker_id: str,
    queue: TaskQueueProtocol,
    handler_registry: dict[str, Callable[Ellipsis, Any]],
    config: TaskWorkerConfig | None = None,
    services: TaskWorkerServices | None = None,
    *,
    middleware_pipeline: TaskMiddlewarePipeline | None = None
)
source

Initialize task worker.

Parameters
ParameterTypeDescription
worker_idstrUnique worker identifier.
queueTaskQueueProtocolTaskQueueProtocol to pull jobs from.
handler_registrydict[str, Callable[Ellipsis, Any]]Dict mapping job names to handler functions.
configTaskWorkerConfig | NoneOptional worker configuration controlling timeouts and execution behaviour. Defaults to TaskWorkerConfig() when not provided.
servicesTaskWorkerServices | NoneOptional bundle of infrastructure services (DLQ, result store, progress store, dashboard, retry policy, task manager, idempotency manager, logger). Use TaskWorkerServices to compose only the services you need; absent services degrade gracefully. Pass a `logger via services.logger` to inject a custom logger; otherwise an unbound module logger is used so worker identifiers are not added to log records.
middleware_pipelineTaskMiddlewarePipeline | NoneOptional middleware pipeline for cross-cutting concerns.
get_stats
def get_stats() -> WorkerJobStats
source

Get worker statistics

Returns
TypeDescription
WorkerJobStatsCurrent worker statistics
is_busy
def is_busy() -> bool
source

Check if worker is currently processing a job

Returns
TypeDescription
boolTrue if worker has a job in progress
get_progress_tracker
def get_progress_tracker(job_id: str) -> ProgressTracker | None
source

Get a ProgressTracker for a job (opt-in for handlers).

Handlers can use this to report progress. The worker must have a progress_store configured.

Parameters
ParameterTypeDescription
job_idstrThe job ID to track progress for
Returns
TypeDescription
ProgressTracker | NoneProgressTracker instance if progress_store is configured, None otherwise

source
Configuration for task workers.

model_config: ClassVar[ConfigDict] = ConfigDict(extra=“ignore”)

source
Optional infrastructure services injected into a TaskWorker.

Groups all optional infrastructure dependencies so that the TaskWorker constructor stays focused on the mandatory parameters. All fields default to None — omitted services degrade gracefully.

Attributes: container: DI container for resolving task dependencies. dead_letter_queue: DLQ for routing permanently-failed jobs. result_store: Persistent store for job execution results. progress_store: Store for in-progress task progress updates. dashboard: Observability dashboard for worker and job metrics. retry_policy: Retry policy implementation (from a resilience extension). task_manager: Kernel task manager for critical shutdown registration. idempotency_manager: Manager for at-most-once job execution. logger: Optional logger instance.

source
Background task workers, scheduling, and async job queues.

Call configure to configure the task-processing subsystem.

Usage (in-memory queue for development)

from quadkit.tasks.backends.memory import MemoryTaskQueue


@module(
    imports=[TasksModule.configure(queue=MemoryTaskQueue())]
)
class AppModule(Module):
    pass
from quadkit.tasks.backends.memory import MemoryTaskQueue


@module(
    imports=[TasksModule.configure(queue=MemoryTaskQueue())]
)
class AppModule(Module):
    pass

Usage (Redis queue for production)

from quadkit.tasks.backends.redis import RedisTaskQueue


@module(
    imports=[
        TasksModule.configure(
            queue=RedisTaskQueue(url="redis://localhost"),
            worker_count=4,
        )
    ]
)
class AppModule(Module):
    pass
from quadkit.tasks.backends.redis import RedisTaskQueue


@module(
    imports=[
        TasksModule.configure(
            queue=RedisTaskQueue(url="redis://localhost"),
            worker_count=4,
        )
    ]
)
class AppModule(Module):
    pass
configure
def configure(
    cls,
    queue: Any | None = None,
    worker_count: int | None = None,
    enable_scheduler: bool | None = None,
    config: TaskConfig | Any | None = None,
    task_modules: list[str] | tuple[str, Ellipsis] | None = None,
    task_packages: list[str] | tuple[str, Ellipsis] | None = None
) -> DynamicModule
source

Create a TasksModule with explicit or YAML-resolved configuration.

Parameters
ParameterTypeDescription
queueAny | NoneOptional explicit TaskQueueProtocol implementation. When `None, the queue is instantiated from config.backend` via TaskBackendRegistry (defaulting to MemoryTaskQueue).
worker_countint | NoneOptional number of concurrent task workers (`0 for producer-only processes). Defaults to config.worker.worker_count`.
enable_schedulerbool | NoneOptional flag enabling the cron scheduler. Defaults to `config.scheduler.enabled`.
configTaskConfig | Any | NoneTaskConfig or `None to resolve the typed tasks` yaml section at registration time.
task_moduleslist[str] | tuple[str, Ellipsis] | NoneExact Python module paths to import during boot and scan for decorated `@task / @scheduled` callables.
task_packageslist[str] | tuple[str, Ellipsis] | NonePackage roots to import recursively during boot and scan for decorated task callables.
Returns
TypeDescription
DynamicModuleA DynamicModule descriptor.
Raises
ExceptionDescription
TypeErrorIf *config* is not a `TaskConfig or None`.
scope
def scope(
    cls,
    *handlers: type,
    **kwargs: Any
) -> DynamicModule
source

Scope task handler classes into a feature module.

Registers the given task handler classes as providers so they are discovered and wired by the task executor. The parent module graph must already include TasksModule.configure — this does not create a new queue or worker pool.

Uses the anonymous token pattern so both configure() and scope() can coexist in the same compiled graph without a ModuleDuplicateError.

Example

@module(
    imports=[
        TasksModule.configure(queue=RedisTaskQueue(url=...)),
        TasksModule.scope(SendEmailTask, GenerateReportTask),
    ]
)
class NotificationFeatureModule(Module):
    pass
@module(
    imports=[
        TasksModule.configure(queue=RedisTaskQueue(url=...)),
        TasksModule.scope(SendEmailTask, GenerateReportTask),
    ]
)
class NotificationFeatureModule(Module):
    pass
Parameters
ParameterTypeDescription
Returns
TypeDescription
DynamicModuleA DynamicModule scoped to this feature.
stub
def stub(
    cls,
    config: Any = None
) -> DynamicModule
source

Return an in-memory TasksModule for unit testing.

Uses a MemoryTaskQueue with a single worker and no job scheduler.

Returns
TypeDescription
DynamicModuleA DynamicModule backed by an in-memory task queue.

source
Enforces execution timeout on tasks.

Uses asyncio.wait_for to enforce timeout on task execution.

__init__
def __init__(
    default_timeout: float = 300.0,
    max_timeout: float | None = None
) -> None
source
before_execute
async def before_execute(ctx: TaskExecutionContext) -> None
source

Compute and store the effective timeout in the execution context metadata.

execute_with_timeout
async def execute_with_timeout(
    handler: Any,
    job: JobProtocol,
    ctx: TaskExecutionContext,
    resolved_args: list[Any] | None = None,
    resolved_kwargs: dict[str, Any] | None = None
) -> JobResult
source

Execute handler with timeout enforcement.

Parameters
ParameterTypeDescription
handlerAnyThe handler to execute
jobJobProtocolThe job being executed
ctxTaskExecutionContextExecution context
resolved_argslist[Any] | NonePre-resolved positional arguments (from DI)
resolved_kwargsdict[str, Any] | NonePre-resolved keyword arguments (from DI)
Returns
TypeDescription
JobResultJobResult from execution

source
Decorator for unique task execution.

Prevents duplicate execution of the same task using in-memory locks.

By default (skip_if_locked=True) a second caller whose lock key is already held returns None immediately. With skip_if_locked=False the second caller blocks until the key is free (up to timeout seconds), then returns None if the key is still held.

Example

lock_manager = LockManager()
@UniqueTask(
    lock_manager=lock_manager,
    key_func=lambda user_id: f"sync_user:{user_id}",
    timeout=3600
)
async def sync_user(user_id: str):
    await sync_user_data(user_id)
__init__
def __init__(
    lock_manager: LockManager,
    key_func: Callable[[Any], str],
    timeout: float = 60.0,
    skip_if_locked: bool = True
)
source

Initialize unique task decorator.

Parameters
ParameterTypeDescription
lock_managerLockManagerLockManager instance for creating locks
key_funcCallable[[Any], str]Function to generate lock key from task args
timeoutfloatLock expiry in seconds; also the maximum time the wait mode (skip_if_locked=False) blocks for the lock
skip_if_lockedboolIf True, skip execution (return None) if the lock is held; if False, wait up to *timeout* seconds for the lock, then skip if it is still held

source
Worker statistics for monitoring performance.

Tracks execution metrics for an individual worker including job counts, timing, and activity tracking.

uptime
property uptime() -> float
source

Get worker uptime in seconds.

record_job
def record_job(result: JobResult) -> None
source

Record job execution statistics.

Parameters
ParameterTypeDescription
resultJobResultJobProtocol execution result.

source
Pool of task workers with monitoring and scaling

WorkerPool manages multiple TaskWorker instances for concurrent job processing. Supports dynamic scaling and aggregate statistics.

Example

pool = WorkerPool(queue, handlers, size=4)
await pool.start()
stats = pool.get_pool_stats()
await pool.scale_to(8)  # Scale up
await pool.stop()
__init__
def __init__(
    queue: TaskQueueProtocol,
    handler_registry: dict[str, Callable[Ellipsis, Any]] | HandlerRegistry,
    size: int = 1,
    logger: Logger | None = None,
    task_manager: TaskManagerProtocol | None = None,
    dead_letter_queue: DeadLetterQueue | None = None,
    hooks: HookRegistryProtocol | None = None,
    container: Any = None,
    middleware_pipeline: TaskMiddlewarePipeline | None = None
)
source

Initialize worker pool

Parameters
ParameterTypeDescription
queueTaskQueueProtocolTaskQueueProtocol for workers to pull from
handler_registrydict[str, Callable[Ellipsis, Any]] | HandlerRegistryDict mapping job names to handlers, or HandlerRegistry for refreshable lookups after late registrations.
sizeintNumber of workers to create
task_managerTaskManagerProtocol | NoneOptional kernel task manager for critical task registration during worker lifecycle.
dead_letter_queueDeadLetterQueue | NoneOptional shared DLQ for permanently failed jobs.
containerAnyOptional DI container for resolving task dependencies at runtime.
middleware_pipelineTaskMiddlewarePipeline | NoneOptional middleware pipeline applied to every worker created by this pool.
refresh_handlers
def refresh_handlers() -> dict[str, Callable[Ellipsis, Any]]
source

Refresh handlers from registry if using dynamic registry.

start
async def start() -> None
source

Start the worker pool

Creates and starts all workers in the pool.

stop
async def stop(
    drain: bool = False,
    timeout: float | None = 30.0
) -> None
source

Stop the worker pool gracefully.

Parameters
ParameterTypeDescription
drainboolIf `True, wait for each worker to finish its current job before stopping. Defaults to False` for a fast cancel-and-stop.
timeoutfloat | NonePer-worker seconds to wait when *drain* is `True. None` means wait indefinitely. Defaults to 30 seconds.
get_worker_stats
def get_worker_stats() -> list[WorkerJobStats]
source

Get statistics for all workers

Returns
TypeDescription
list[WorkerJobStats]List of WorkerJobStats for each worker
get_pool_stats
def get_pool_stats() -> dict[str, Any]
source

Get aggregate pool statistics

Returns
TypeDescription
dict[str, Any]Dictionary with pool-wide statistics
scale_to
async def scale_to(new_size: int) -> None
source

Scale the worker pool to a new size

Dynamically add or remove workers.

Parameters
ParameterTypeDescription
new_sizeintTarget number of workers

source
Result from a complete workflow execution.

source
Enumeration of possible workflow execution states.

Functions

Section titled “Functions”

chain
def chain(
    *steps: TaskStep,
    stop_on_error: bool = True
) -> TaskChain
source
Create a sequential TaskChain from the given steps.

Each step receives the output of the previous step as its input.

Parameters
ParameterTypeDescription
stop_on_errorboolWhether to abort the pipeline on the first failure. Defaults to `True`.
Returns
TypeDescription
TaskChainA configured TaskChain ready to `await .execute()`.

Example

result = await chain(
    TaskStep("validate", validate_input),
    TaskStep("process", process_data),
    TaskStep("persist", save_to_db),
).execute(raw_data)
result = await chain(
    TaskStep("validate", validate_input),
    TaskStep("process", process_data),
    TaskStep("persist", save_to_db),
).execute(raw_data)

create_memory_task_provider
def create_memory_task_provider(
    worker_count: int = 1,
    enable_scheduler: bool = True
) -> TaskProvider
source
Create a TaskProvider backed by an in-memory queue.
Parameters
ParameterTypeDescription
worker_countintNumber of worker coroutines to run concurrently.
enable_schedulerboolWhether to enable the cron job scheduler.
Returns
TypeDescription
TaskProviderTaskProvider with MemoryTaskQueue.

create_provider_from_config
def create_provider_from_config(config: Any) -> TaskProvider
source
Create a TaskProvider from a TaskConfig instance or dict.
Parameters
ParameterTypeDescription
configAnyTaskConfig instance or a dict that will be coerced into one.
Returns
TypeDescription
TaskProviderFully configured TaskProvider.

create_rabbitmq_task_provider
def create_rabbitmq_task_provider(
    amqp_url: str = 'amqp://localhost:5672/',
    queue_name: str = 'tasks',
    worker_count: int = 1,
    enable_scheduler: bool = True
) -> TaskProvider
source
Create a TaskProvider backed by a RabbitMQ queue.
Parameters
ParameterTypeDescription
amqp_urlstrAMQP connection URL.
queue_namestrName of the AMQP queue.
worker_countintNumber of worker coroutines to run concurrently.
enable_schedulerboolWhether to enable the cron job scheduler.
Returns
TypeDescription
TaskProviderTaskProvider with RabbitMQTaskQueue.

create_redis_task_provider
def create_redis_task_provider(
    redis_url: str = 'redis://localhost:6379',
    queue_name: str = 'tasks',
    worker_count: int = 1,
    enable_scheduler: bool = True
) -> TaskProvider
source
Create a TaskProvider backed by a Redis queue.
Parameters
ParameterTypeDescription
redis_urlstrRedis connection URL.
queue_namestrName of the Redis list used as the task queue.
worker_countintNumber of worker coroutines to run concurrently.
enable_schedulerboolWhether to enable the cron job scheduler.
Returns
TypeDescription
TaskProviderTaskProvider with RedisTaskQueue.

delay
def delay(fn: Callable[_P, Coroutine[Any, Any, _R]]) -> _DelayedCallable[_P, _R]
source
Decorate an async function to add a ``.delay()`` background-dispatch method.

The decorated function is unchanged when called directly — it still returns a coroutine that must be await-ed. The .delay() method is the only addition; it schedules the coroutine as a background asyncio.Task and stores the reference to prevent premature garbage collection.

Parameters
ParameterTypeDescription
fnCallable[_P, Coroutine[Any, Any, _R]]An async function (coroutine function) to wrap.
Returns
TypeDescription
_DelayedCallable[_P, _R]A _DelayedCallable that is callable exactly like `fn but also exposes .delay(*args, **kwargs) -> asyncio.Task`.

Example

@delay
async def flush_audit_log(batch: list[AuditEntry]) -> None:
    await audit_store.save_many(batch)


# Direct call (awaited in the current task):
await flush_audit_log(batch=entries)


# Background dispatch (does not block):
flush_audit_log.delay(batch=entries)

distributed_lock
async def distributed_lock(
    key: str,
    timeout: float = 60.0,
    lock_manager: LockManager | None = None
) -> AsyncGenerator[InMemoryLock, None]
source
Context manager for in-memory locks.

Deprecated: Use LockManager.acquire() instead for better DI integration.

Parameters
ParameterTypeDescription
keystrUnique lock key
timeoutfloatLock timeout in seconds
lock_managerLockManager | NoneOptional LockManager instance (creates new if None)
Yields
TypeDescription
AsyncGenerator[InMemoryLock, None]InMemoryLock instance

Example

manager = LockManager()  # Should be from DI container
async with distributed_lock("resource:123", lock_manager=manager):
    await process_resource(123)

scheduled
def scheduled(
    cron: str,
    name: str | None = None,
    priority: Priority | int = Priority.NORMAL,
    max_retries: int = 3,
    timeout: float | None = None,
    queue: str = 'default'
) -> Callable[[Any], Any]
source
Decorator to define a scheduled task (function or ``ScheduledTask`` class).
Parameters
ParameterTypeDescription
cronstrCron expression (e.g., "0 9 * * *")
namestr | NoneTask name
priorityPriority | intTask priority
max_retriesintMaximum retry attempts
timeoutfloat | NoneExecution timeout
queuestrQueue name for this task
Returns
TypeDescription
Callable[[Any], Any]Decorated class or task wrapper with scheduling metadata.

task
def task(
    name: str | None = None,
    priority: Priority | int = Priority.NORMAL,
    max_retries: int = 3,
    timeout: float | None = None,
    queue: str = 'default',
    idempotency_key: str | None = None
) -> Callable[[Any], Any]
source
Decorator to define a task.

Supports both async/sync functions and class-based tasks defining an execute(self, ...) method (such as ScheduledTask subclasses).

Parameters
ParameterTypeDescription
namestr | NoneTask name (defaults to function/class name)
priorityPriority | intTask priority (HIGH, NORMAL, LOW)
max_retriesintMaximum retry attempts on failure
timeoutfloat | NoneExecution timeout in seconds
queuestrQueue name for this task
idempotency_keystr | NoneOptional key for idempotent execution
Returns
TypeDescription
Callable[[Any], Any]Decorated class (preserving class identity) or `TaskFunction` wrapper.

Exceptions

Section titled “Exceptions”

source
The queue has reached its maximum capacity and cannot accept new tasks.

The caller should back off and retry after a delay, or apply backpressure.

__init__
def __init__(
    message: str = 'Task queue is full',
    *,
    queue_name: str | None = None,
    capacity: int | None = None,
    **kwargs: Any
) -> None
source

source
Raised when a task is cancelled

source
Base exception for all task-related errors

source
Raised when a task execution fails
__init__
def __init__(
    message: str,
    original_error: Exception | None = None,
    details: dict[str, Any] | None = None,
    hint: str | None = None
) -> None
source

source
Raised when a task cannot be found

source
Raised when a task execution exceeds timeout

source
Raised when task parameters are invalid