task
API reference for task.
Task definitions and enqueueing.
class BoundTask
A task with per-call options, created by :meth:Task.options.
__init__(task: Task[P, R], call: CallOptions) -> None
enqueue(args: P.args = (), kwargs: P.kwargs = {}) -> TaskHandle[R]
enqueue_many(items: Iterable[tuple[Any, ...]]) -> list[TaskHandle[R]]
Publish many calls with these options in one pipelined round-trip.
enqueue_many_sync(items: Iterable[tuple[Any, ...]]) -> list[TaskHandle[R]]
enqueue_sync(args: P.args = (), kwargs: P.kwargs = {}) -> TaskHandle[R]
class CallOptions
__init__(queue: str | None = None, eta: float | None = None, task_id: str | None = None, correlation_id: str | None = None, headers: Mapping[str, str] | None = None, timeout: float | None = None, priority: Priority | None = None) -> None
class ConfigurationError
Invalid configuration or API misuse.
class Envelope
The on-the-wire task message.
Encoded positionally (array_like) for compactness; new fields must be
appended with defaults so older messages keep decoding.
class PublishRequest
__init__(queue: str, task_id: str, data: bytes, eta: float | None = None, record: bytes | None = None, record_ttl: int | None = None) -> None
class Queue
A BlitzQ application.
Queue holds configuration and the task registry, and provides client
operations (enqueue, results, inspection, cancellation). The same object
is imported by web processes, which only publish, and by workers started
with blitzq worker module:attribute.
Parameters
namestr— Default queue for tasks without an explicit queue or routing rule.redis_urlstr | None— Redis connection URL (rediss://for TLS). Defaults to theBLITZQ_REDIS_URLenvironment variable, thenredis://localhost:6379/0.modeLiteral['fast', 'reliable']—"reliable"(default; Redis Streams, at-least-once, crash recovery) or"fast"(Redis lists, at-most-once, lowest overhead).brokerBroker | None— A custom :class:~blitzq.broker.base.Brokerinstead of Redis.store_resultsbool— Store final state and return values (can be overridden per task).result_ttlint— Seconds to keep task records.track_statebool— Also recordqueued/scheduled/running/retryingtransitions. Costs one extra write per transition.routesRoutes— Task-name glob patterns to queue names, or a callable.visibility_timeoutfloat— Reliable mode: seconds after which an unacknowledged message whose worker stopped renewing its lease is redelivered.max_deliveriesint— Reliable mode: dead-letter a message after this many deliveries (protects against messages that crash workers).
__init__(name: str = 'default', redis_url: str | None = None, mode: Literal['fast', 'reliable'] = 'reliable', broker: Broker | None = None, namespace: str = 'blitzq', serializer: Serializer | None = None, store_results: bool = True, result_ttl: int = 24 * 3600, track_state: bool = False, routes: Routes = None, default_retries: int = 0, default_retry_policy: RetryPolicy | None = None, default_timeout: float | None = None, visibility_timeout: float = 60.0, max_deliveries: int = 5, dead_letter_max: int = 100000, revoke_ttl: int = 3600, redis_options: dict[str, Any] | None = None) -> None
add_sync_wrapper(wrapper: SyncWrapper) -> None
Wrap every sync-task call inside its executor thread.
wrapper(call) must invoke call() and return its result. Used by
integrations that need thread-local setup, such as Django connection
management or a Flask application context.
cancel(task_id: str) -> bool
Cancel a task.
Returns True when the task was still scheduled (delayed or waiting
for a retry) and has been removed - cancellation is then certain.
Otherwise a revocation is recorded and False is returned: workers
skip the task if they receive it after their next revocation sync
(about one second) and cancel it if it is a running async task.
Running thread/process tasks cannot be interrupted, and a task that
already finished is unaffected.
cancel_sync(task_id: str) -> bool
close() -> None
Close async broker connections. Safe to call more than once.
close_sync() -> None
Close connections used by the synchronous API and stop its thread.
connect() -> None
Eagerly connect the async broker (optional; connections are lazy).
dead_letters(limit: int = 100, offset: int = 0) -> list[DeadLetter]
get_result(task_id: str, timeout: float | None = None) -> Any
Wait for a task's final state and return its result.
Waits on the broker's push notification when it has one (Redis:
pub/sub on the task's record channel), falling back to a coarse
poll otherwise. Requires result storage for the task. Raises
TaskFailed for failed, dead-lettered or cancelled tasks and
ResultTimeout after timeout seconds (None waits
indefinitely).
get_result_sync(task_id: str, timeout: float | None = None) -> Any
inspect(task_id: str) -> TaskInfo | None
Everything known about a task, or None.
None means the task is queued or running without state tracking,
its record expired, or the id is unknown.
inspect_sync(task_id: str) -> TaskInfo | None
new_task_id() -> str
on_shutdown(fn: Hook) -> Hook
Register a worker/scheduler shutdown hook (sync or async).
on_startup(fn: Hook) -> Hook
Register a worker/scheduler startup hook (sync or async).
periodic(schedule: Schedule | str | float | timedelta, tz: str | None = None, args: Sequence[Any] = (), kwargs: dict[str, Any] | None = None, missed: MissedPolicy = 'run_once', name: str | None = None, queue: str | None = None, task_options: Any = {}) -> Callable[[Callable[..., Any]], Task[Any, Any]]
Register a periodic task (dispatched by blitzq scheduler).
schedule is a cron string ("*/5 * * * *", evaluated in tz),
an interval in seconds or a timedelta, or a :class:Schedule.
missed controls what happens to occurrences missed while no
scheduler was running: "run_once" dispatches only the latest,
"run_all" dispatches each (up to 100), "skip" dispatches none.
purge(queue: str | None = None) -> int
Delete waiting messages from queue (default queue if omitted).
queue_stats(queues: Sequence[str] | None = None) -> list[QueueStats]
retry(task_id: str) -> bool
Re-enqueue a dead-lettered task with a fresh attempt budget.
Returns False if no dead letter with this id exists.
retry_sync(task_id: str) -> bool
send(task_name: str, args: Sequence[Any] = (), kwargs: dict[str, Any] | None = None, options: Any = {}) -> TaskHandle[Any]
Enqueue a task by name, without importing its code (producer-only apps).
send_sync(task_name: str, args: Sequence[Any] = (), kwargs: dict[str, Any] | None = None, options: Any = {}) -> TaskHandle[Any]
status(task_id: str) -> TaskState | None
status_sync(task_id: str) -> TaskState | None
task(fn: Callable[..., Any] | None = None, name: str | None = None, queue: str | None = None, retries: int | None = None, retry_policy: RetryPolicy | None = None, timeout: float | None = None, executor: Executor | None = None, store_result: bool | None = None, dead_letter: bool = True, priority: Priority = 'normal', rate_limit: RateLimit | str | None = None) -> Any
Register a task.
retries is the number of retries after the first attempt.
executor defaults to "async" for coroutine functions and
"thread" for regular functions; use "process" for CPU-bound
functions (they must be importable module-level functions).
dead_letter=False records terminal failures as failed instead
of moving them to the dead-letter store.
priority ("high"/"normal"/"low") is this task's
default priority within its queue; override per call with
task.options(priority=...). Priority levels are physically
separate broker queues that every worker checks in order (high,
then normal, then low) while sharing the queue's overall concurrency
- not a separate queue you need to remember to subscribe a worker to. See docs/architecture.md#task-priority.
rate_limit caps how often this task starts execution, shared
across every worker (a Redis-backed token bucket, not a per-process
counter): "10/s", "100/m" or "1000/hour", or a
:class:~blitzq.RateLimit. A task over its limit is not executed and
not counted as a retry; it is rescheduled for when a slot should be
free. See docs/architecture.md#rate-limiting.
class RateLimit
A cap of count task starts per period_seconds, enforced globally.
The bucket holds up to count tokens (one burst's worth) and refills
continuously at count / period_seconds tokens/second, so a limit of
"100/m" allows a burst of 100 immediately after being idle, then
settles to roughly one every 0.6s - it does not release exactly 100 once
a minute.
__init__(count: float, period_seconds: float) -> None
parse(spec: str) -> RateLimit
Parse "100", "100/s", "100/m" or "100/hour".
class RetryPolicy
Exponential backoff with optional jitter.
The delay before retry n (n = 1 for the first retry) is
min(max_delay, initial_delay * backoff ** (n - 1)). With jitter
enabled the delay is drawn uniformly from [delay / 2, delay] ("equal
jitter"), which spreads retries of simultaneously failing tasks while
keeping a guaranteed minimum wait.
retry_on lists exception types that trigger an automatic retry;
dont_retry_on takes precedence and marks exceptions as permanent
failures. An explicit :class:~blitzq.Retry raised by the task is always
honoured while attempts remain.
__init__(initial_delay: float = 1.0, max_delay: float = 300.0, backoff: float = 2.0, jitter: bool = True, retry_on: tuple[type[BaseException], ...] = (Exception,), dont_retry_on: tuple[type[BaseException], ...] = ()) -> None
compute_delay(retry_number: int, rng: random.Random | None = None) -> float
Delay in seconds before retry number retry_number (1-based).
is_retryable(exc: BaseException) -> bool
class Task
A function registered with a :class:~blitzq.Queue.
Calling the object runs the function directly in the current process.
enqueue publishes it for a worker. For async def functions R is
the awaited return type.
__init__(app: Queue, fn: Callable[P, Any], options: TaskOptions) -> None
enqueue(args: P.args = (), kwargs: P.kwargs = {}) -> TaskHandle[R]
Publish the task. Non-blocking for the event loop; returns once Redis accepted it.
enqueue_many(items: Iterable[tuple[Any, ...]]) -> list[TaskHandle[R]]
Publish many calls in one pipelined round-trip.
Each item is a tuple of positional arguments.
enqueue_many_sync(items: Iterable[tuple[Any, ...]]) -> list[TaskHandle[R]]
enqueue_sync(args: P.args = (), kwargs: P.kwargs = {}) -> TaskHandle[R]
Blocking variant of :meth:enqueue for synchronous code.
options(queue: str | None = None, delay: float | timedelta | None = None, eta: datetime | float | None = None, task_id: str | None = None, correlation_id: str | None = None, headers: Mapping[str, str] | None = None, timeout: float | None = None, priority: Priority | None = None) -> BoundTask[P, R]
Return a view of this task with per-call options applied.
delay (seconds or timedelta) and eta (aware datetime or epoch
seconds) schedule the task for later. task_id sets an explicit id;
enqueueing the same id twice creates two messages unless the first is
still scheduled, in which case its schedule entry is replaced.
priority ("high"/"normal"/"low") reorders this call
within its queue's shared concurrency budget; see the priority
parameter of @queue.task for how workers pick it up.
class TaskHandle
Reference to an enqueued task.
Async methods (result, info, status, cancel) must be awaited
inside an event loop; *_sync variants block the calling thread and must
not be used inside a running event loop.
__init__(task_id: str, app: Queue, queue: str) -> None
cancel() -> bool
cancel_sync() -> bool
info() -> TaskInfo | None
result(timeout: float | None = None) -> R
Wait for the task to finish and return its result.
Raises :class:~blitzq.exceptions.TaskFailed if it failed, was
dead-lettered or cancelled, and :class:~blitzq.exceptions.ResultTimeout
if no final state is visible within timeout seconds.
result_sync(timeout: float | None = None) -> R
status() -> TaskState | None
status_sync() -> TaskState | None
class TaskInfo
A snapshot of what BlitzQ knows about a task.
Records are written by producers (queued/scheduled when state
tracking is enabled) and by workers (running when tracking is enabled,
and the final state when results are stored). Readers may observe a
slightly stale state during concurrent transitions; see
docs/delivery_guarantees.md.
class TaskOptions
__init__(name: str, queue: str | None, max_attempts: int, retry_policy: RetryPolicy, timeout: float | None, executor: Executor, store_result: bool, dead_letter: bool, priority: Priority, rate_limit: RateLimit | None) -> None
class TaskState
current_task() -> TaskContext | None
Return the context of the task being executed, or None outside a task.