API reference¶

App and tasks¶

Registry of tasks plus the producer and consumer that move their messages.

Attributes:

Name	Type	Description
`name`		Name of the app.
`producer`		Used by `Task.delay`; None if the app only consumes.
`consumer`		Used by `Worker`; None if the app only produces.
`store`	`Store`	Key-value store for task states, task results and scheduler claim keys.
`codec`	`Codec`	Codec that encodes messages, task states and results.
`tz`		Default timezone for scheduled tasks.
`options`		Default options of the app's tasks.
`hooks`		The hooks and middleware registered through the `before_`, `after_` and `middleware` decorators.
`telemetry`	`Telemetry`	Tracing and metrics component, chosen when the app is built: the OpenTelemetry one when `opentelemetry-api` is installed, else a `NullTelemetry`.

`scheduled_tasks` `property` ¶

The registered scheduled tasks, in registration order.

`tasks` `property` ¶

The registered tasks, by name, as a read-only mapping.

`init(name, *, producer=None, consumer=None, store=None, codec=None, tz='UTC', options=None)` ¶

Initializes the app.

Parameters:

Name	Type	Description	Default
`name`	`str`	Name of the app.	required
`producer`	`Producer \| None`	Used by `Task.delay`; None if the app only consumes.	`None`
`consumer`	`Consumer \| None`	Used by `Worker`; None if the app only produces.	`None`
`store`	`Store \| None`	Key-value store for task states, task results and scheduler claim keys; defaults to a `NullStore`.	`None`
`codec`	`Codec \| None`	Codec that encodes messages, task states and results; defaults to a `JsonCodec`.	`None`
`tz`	`str`	Default timezone for scheduled tasks.	`'UTC'`
`options`	`TaskOptions \| None`	Default options for the app's tasks.	`None`

Raises:

Type	Description
`TypeError`	If `options` is not a `TaskOptions`.
`ValueError`	If the timezone is unknown.

`after_shutdown(hook)` ¶

Registers a hook run after a worker or scheduler has stopped, before the store stops.

Same rules as before_startup.

Parameters:

Name	Type	Description	Default
`hook`	`LifecycleHook`	The hook to register.	required

Returns:

Type	Description
`LifecycleHook`	The hook, unchanged.

`after_startup(hook)` ¶

Registers a hook run once a worker or scheduler has started.

Same rules as before_startup.

Parameters:

Name	Type	Description	Default
`hook`	`LifecycleHook`	The hook to register.	required

Returns:

Type	Description
`LifecycleHook`	The hook, unchanged.

`before_shutdown(hook)` ¶

Registers a hook run when a worker or scheduler begins to shut down.

A worker runs it when it stops fetching, before draining. Same rules as before_startup.

Parameters:

Name	Type	Description	Default
`hook`	`LifecycleHook`	The hook to register.	required

Returns:

Type	Description
`LifecycleHook`	The hook, unchanged.

`before_startup(hook)` ¶

Registers a hook run after the store starts and before serving begins.

It runs before a worker's consumer or a scheduler's loop starts. Lifecycle hooks may be sync or async and receive the Runner they run in, a Worker or a Scheduler; runner.role tells them apart. Registering the same hook twice has no effect.

Example

Initialize telemetry in every process, but open a database pool only in workers:

@app.before_startup
async def startup(runner: Runner) -> None:
    init_sentry()
    if runner.role == "worker":
        await database.connect()

Parameters:

Name	Type	Description	Default
`hook`	`LifecycleHook`	The hook to register.	required

Returns:

Type	Description
`LifecycleHook`	The hook, unchanged.

`get_task(task_id)` ¶

Returns a handle for reading the state and result of a published task.

Parameters:

Name	Type	Description	Default
`task_id`	`str`	The task's id (`Envelope.id`).	required

Returns:

Type	Description
`TaskHandle[Any]`	A `TaskHandle` reading this app's store with this app's codec.

`middleware(layer)` ¶

Registers a middleware layer around each task.

A layer is an async function (call: TaskCall, next: Next) -> TaskResult. It receives the call and next, which runs the rest of the chain and the task, and returns a TaskResult: the one from next or another one, such as call.skip(), the SKIPPED result that vetoes the task. The last registered layer is the outermost. Registering the same layer twice has no effect.

Parameters:

Name	Type	Description	Default
`layer`	`Middleware`	The middleware layer to register.	required

Returns:

Type	Description
`Middleware`	The layer, unchanged.

Example

Log how long each task takes:

@app.middleware
async def timed(call: TaskCall, next: Next) -> TaskResult:
    started = time.monotonic()
    result = await next(call)
    logger.info("%s took %.3fs", call.envelope.task, time.monotonic() - started)
    return result

`publish(envelope, options=None)` `async` ¶

Publishes an envelope for a worker to run.

The telemetry component first adds trace context to the envelope's metadata (keys already there are kept). After the envelope is sent, the task's state is set to PENDING unless a worker has already recorded one. A store failure is logged and does not fail the publish.

Parameters:

Name	Type	Description	Default
`envelope`	`Envelope`	The envelope to publish.	required
`options`	`PublishOptions \| None`	Publish options such as a delay.	`None`

Returns:

Type	Description
`TaskHandle[Any]`	A `TaskHandle` whose `task_id` is the envelope id and whose `receipt` is
`TaskHandle[Any]`	the producer's receipt.

Raises:

Type	Description
`NotConfiguredError`	If the app has no producer.
`EncodeError`	If the codec cannot encode the envelope.

`registered_task(name)` ¶

Looks up a registered task.

Parameters:

Name	Type	Description	Default
`name`	`str`	The task name.	required

Returns:

Type	Description
`Task[..., Any]`	The task.

Raises:

Type	Description
`KeyError`	If no task has that name.

`scheduled_task(*, cron, tz=None, args=(), kwargs=None, name=None, options=None)` ¶

Decorator that registers a function as a task published on a cron schedule.

Parameters:

Name	Type	Description	Default
`cron`	`str`	Five-field cron expression.	required
`tz`	`str \| None`	IANA timezone for the expression; defaults to the app's.	`None`
`args`	`Sequence[object]`	Positional arguments for each published call.	`()`
`kwargs`	`Mapping[str, object] \| None`	Keyword arguments for each published call.	`None`
`name`	`str \| None`	Task name; defaults to the app name and the function's qualified name.	`None`
`options`	`TaskOptions \| None`	Options of this task; they override the app's defaults.	`None`

Returns:

Type	Description
`Callable[[Callable[P, Awaitable[R]]], ScheduledTask[P, R]]`	A decorator producing a `ScheduledTask`.

Raises:

Type	Description
`CronError`	If the cron expression is invalid.
`TypeError`	If `options` is not a `TaskOptions`.
`ValueError`	If the timezone is unknown, or the name is already registered.

`send(envelope, options=None)` `async` ¶

Encodes an envelope and publishes it, without recording any state.

Parameters:

Name	Type	Description	Default
`envelope`	`Envelope`	The envelope to send.	required
`options`	`PublishOptions \| None`	Publish options such as a delay.	`None`

Returns:

Type	Description
`Any`	The producer's receipt.

Raises:

Type	Description
`NotConfiguredError`	If the app has no producer.
`EncodeError`	If the codec cannot encode the envelope.

`task(*, name=None, options=None)` ¶

Decorator that registers a function as a task.

Parameters:

Name	Type	Description	Default
`name`	`str \| None`	Task name; defaults to the app name and the function's qualified name.	`None`
`options`	`TaskOptions \| None`	Options of this task; they override the app's defaults.	`None`

Returns:

Type	Description
`Callable[[Callable[P, Awaitable[R]]], Task[P, R]]`	A decorator producing a `Task`.

Raises:

Type	Description
`TypeError`	If `options` is not a `TaskOptions`.
`ValueError`	If the name is already registered.

Bases: Generic[P, R]

A registered async function that can be called directly or enqueued with delay.

Attributes:

Name	Type	Description
`func`		The wrapped `async def` function.
`name`		Unique name used to route messages to this task.
`retry`		The task's retry policy; None for no retries.
`timeout`		Seconds one attempt may run; None for no limit.

`call(*args, **kwargs)` ¶

Calls the wrapped function in-process; await the result.

`delay(*args, **kwargs)` `async` ¶

Publishes a call to this task for a worker to run.

Parameters:

Name	Type	Description	Default
`*args`	`args`	Positional arguments for the task.	`()`
`**kwargs`	`kwargs`	Keyword arguments for the task.	`{}`

Returns:

Type	Description
`TaskHandle[R]`	A `TaskHandle` whose `task_id` is the envelope id and whose `receipt` is
`TaskHandle[R]`	the producer's receipt.

Raises:

Type	Description
`NotConfiguredError`	If the app has no producer.
`EncodeError`	If the app's codec cannot encode an argument.

`with_metadata(**metadata)` ¶

Returns a publisher that attaches metadata to the envelopes it sends.

Parameters:

Name	Type	Description	Default
`**metadata`	`object`	Data carried alongside the call.	`{}`

Returns:

Type	Description
`BoundTask[P, R]`	A `BoundTask` carrying the metadata; this task is left unchanged.

Bases: Task[P, R]

A task that a Scheduler publishes on a cron schedule.

Attributes:

Name	Type	Description
`schedule`		The scheduled task's schedule.

Options of a task, or the defaults of every task of an app.

Attributes:

Name	Type	Description
`retry`	`RetryPolicy \| Literal[False] \| None`	The retry policy; `False` disables retries and `None` leaves the option unset.
`timeout`	`float`	Seconds one attempt of the task may run before it is cancelled and fails with `TaskTimeoutError`; `0` (the default) means no limit.

Raises:

Type	Description
`TypeError`	If `retry` is not a `RetryPolicy`, `False` or `None`, or `timeout` is not a number.
`ValueError`	If `timeout` is negative or not finite.

`policy` `property` ¶

The retry policy in force; None when retries are off or unset.

`with_defaults(defaults)` ¶

Fills the options left unset from defaults.

Parameters:

Name	Type	Description	Default
`defaults`	`TaskOptions`	The options to fall back on.	required

Returns:

Type	Description
`TaskOptions`	New options; neither `self` nor `defaults` is changed.

Bases: Runner

Publishes an app's scheduled tasks when their cron slots come due.

run publishes due slots until shutdown is requested. Each scheduled task starts at its first slot after the current time.

Each publish carries the metadata idempotency_key, "<task name>:<slot>", and scheduled_at, the slot as a UTC ISO-8601 instant.

Attributes:

Name	Type	Description
`app`		The app whose scheduled tasks are published.
`clock`	`Clock`	Source of time and sleeping.
`probe_dir`		Directory for the liveness probe file; None disables it.
`heartbeat_interval`		Seconds between probe file updates.
`grace`		Seconds a slot may be late and still fire.
`role`	`Role`	Always "scheduler".

`init(app, *, clock=None, probe_dir=None, heartbeat_interval=5.0, grace=30.0)` ¶

Initializes the scheduler.

Parameters:

Name	Type	Description	Default
`app`	`App`	The app to publish for; it must have a producer and scheduled tasks.	required
`clock`	`Clock \| None`	Source of time and sleeping; defaults to `SystemClock`.	`None`
`probe_dir`	`Path \| None`	Directory for the liveness probe file; None disables it.	`None`
`heartbeat_interval`	`float`	Seconds between probe file updates.	`5.0`
`grace`	`float`	Seconds a slot may be late and still fire.	`30.0`

Raises:

Type	Description
`NotConfiguredError`	If the app has no producer.
`EncodeError`	If the app's codec cannot encode a scheduled task's arguments.
`ValueError`	If the app has no scheduled tasks, `heartbeat_interval` <= 0, or `grace` is not in [0, 60).

`request_shutdown()` ¶

Asks run to stop; idempotent and safe in signal handlers.

`run()` `async` ¶

Serves the app until shutdown is requested.

The app's store is started before the startup hooks and stopped after the shutdown hooks; see App.before_startup and its siblings.

`run_with_signals()` `async` ¶

Runs like run, with SIGTERM and SIGINT requesting a graceful shutdown.

Running¶

Bases: Runner

Runs an app's tasks, consuming deliveries with bounded concurrency.

run serves until shutdown is requested or deliveries end. In-flight tasks are given shutdown_timeout seconds to finish; stragglers are cancelled and their deliveries requeued.

Attributes:

Name	Type	Description
`app`		The app to serve.
`consumer`	`Consumer`	The app's consumer.
`concurrency`		Maximum tasks running at once.
`shutdown_timeout`		Seconds to let running tasks finish after shutdown is requested.
`probe_dir`		Directory for probe files; None disables them.
`heartbeat_interval`		Seconds between probe file updates.
`role`	`Role`	Always "worker".

`init(app, *, concurrency=10, shutdown_timeout=10.0, probe_dir=None, heartbeat_interval=5.0)` ¶

Initializes the worker.

Parameters:

Name	Type	Description	Default
`app`	`App`	The app to serve; it must have a consumer.	required
`concurrency`	`int`	Maximum tasks running at once.	`10`
`shutdown_timeout`	`float`	Seconds to let running tasks finish after shutdown is requested.	`10.0`
`probe_dir`	`Path \| None`	Directory for probe files; None disables them.	`None`
`heartbeat_interval`	`float`	Seconds between probe file updates.	`5.0`

Raises:

Type	Description
`ValueError`	If `concurrency` < 1, `heartbeat_interval` <= 0 or `shutdown_timeout` < 0.
`NotConfiguredError`	If the app has no consumer.

`handle(delivery)` `async` ¶

Handles one delivery: runs its task and records the outcome.

Parameters:

Name	Type	Description	Default
`delivery`	`Delivery`	The delivery to handle.	required

Returns:

Type	Description
`TaskResult \| None`	The task's result, or None if the body is not a valid envelope.

`request_shutdown()` ¶

Asks run to stop; idempotent and safe in signal handlers.

`run()` `async` ¶

Serves the app until shutdown is requested.

The app's store is started before the startup hooks and stopped after the shutdown hooks; see App.before_startup and its siblings.

`run_with_signals()` `async` ¶

Runs like run, with SIGTERM and SIGINT requesting a graceful shutdown.

Bases: ABC

An asynk process: a Worker or a Scheduler.

Lifecycle hooks receive the runner they run in, so one hook can serve every asynk process and branch on role where it must differ.

Subclasses implement _serve and may override _start and _on_shutdown_requested.

Attributes:

Name	Type	Description
`app`		The app being served.
`probe_dir`		Directory for probe files; None disables them.
`heartbeat_interval`		Seconds between probe file updates.
`role`	`Role`	Which kind of process this is, "worker" or "scheduler".

`init(app, *, probe_dir=None, heartbeat_interval=5.0)` ¶

Initializes the runner.

Parameters:

Name	Type	Description	Default
`app`	`App`	The app to serve.	required
`probe_dir`	`Path \| None`	Directory for probe files; None disables them.	`None`
`heartbeat_interval`	`float`	Seconds between probe file updates.	`5.0`

Raises:

Type	Description
`ValueError`	If `heartbeat_interval` <= 0.

`request_shutdown()` ¶

Asks run to stop; idempotent and safe in signal handlers.

`run()` `async` ¶

Serves the app until shutdown is requested.

The app's store is started before the startup hooks and stopped after the shutdown hooks; see App.before_startup and its siblings.

`run_with_signals()` `async` ¶

Runs like run, with SIGTERM and SIGINT requesting a graceful shutdown.

Middleware¶

One task about to run, as passed through the middleware chain.

Attributes:

Name	Type	Description
`envelope`	`Envelope`	The envelope of the message being handled.
`delivery`	`Delivery`	The delivery the envelope arrived on.

`skip()` ¶

Builds the SKIPPED result.

Return it without calling next to veto the task.

Returns:

Type	Description
`TaskResult`	A `TaskResult` with status `TaskStatus.SKIPPED`.

Calls the rest of the middleware chain, ending with the task.

Retries¶

How a failing task is retried.

Attributes:

Name	Type	Description
`max_retries`	`int \| None`	How many times a failed task is retried; `None` retries without limit and `0` never retries.
`backoff`	`float \| Backoff \| None`	Seconds to wait before each retry, or a `Backoff` strategy; `None` retries immediately.
`retry_on`	`tuple[type[Exception], ...]`	Exception types that trigger a retry; any other error fails the task at once.

Raises:

Type	Description
`TypeError`	If a field has the wrong type.
`ValueError`	If a field is out of range, or `retry_on` is empty.

Exponential backoff: the delay grows by factor on each retry.

The first retry waits base, the next base * factor, and so on, never more than max.

Attributes:

Name	Type	Description
`base`	`float`	Seconds to wait before the first retry; must be positive.
`factor`	`float`	Multiplier applied for each further retry; at least 1.
`max`	`float`	Upper bound of the delay in seconds; at least `base`.
`jitter`	`bool`	When true, the delay is drawn uniformly between zero and the capped delay.

Raises:

Type	Description
`ValueError`	If `base`, `factor` or `max` is out of range.

`delay(retry)` ¶

Returns the seconds to wait before a retry.

Parameters:

Name	Type	Description	Default
`retry`	`int`	The retry number, counted from 1.	required

Returns:

Type	Description
`float`	The delay in seconds, at most `max`.

Bases: Protocol

A strategy that decides how long to wait before a retry.

`delay(retry)` ¶

Returns the seconds to wait before a retry.

Parameters:

Name	Type	Description	Default
`retry`	`int`	The retry number, counted from 1.	required

Returns:

Type	Description
`float`	The delay in seconds; a value of zero or less means no delay.

Transport protocols¶

Bases: Protocol

Sends encoded messages to a broker.

`publish(envelope, body, options=None)` `async` ¶

Publishes one message.

Parameters:

Name	Type	Description	Default
`envelope`	`Envelope`	The envelope the message carries, for routing and broker metadata such as its id.	required
`body`	`bytes`	The encoded message to send.	required
`options`	`PublishOptions \| None`	Publish options such as a delay; a producer applies the ones its broker supports.	`None`

Returns:

Type	Description
`Any`	An implementation-defined receipt (for example a broker message id).

Instructions for publishing one message.

Attributes:

Name	Type	Description
`delay`	`float \| None`	Seconds the broker holds the message before delivering it; None delivers it now.

`__post_init__()` ¶

Validates the options.

Raises:

Type	Description
`TypeError`	If `delay` is not a number.
`ValueError`	If `delay` is negative, NaN or infinite.

Bases: Protocol

Receives deliveries from a broker and decides how each is settled.

The worker calls start, iterates deliveries, closes the iterator, then calls stop. Each delivery is handled inside process(delivery). When a task failed, was unknown or exceeded its retries and the delivery is still unsettled, the worker raises DeliveryFailedError inside that block. On exit, process settles a delivery that is still unsettled (usually ack on a clean exit, reject on an exception) and re-raises the exception.

`connected` `property` ¶

Whether the broker connection is healthy; drives the readiness probe.

`deliveries()` ¶

Returns the stream of incoming deliveries, one per received message.

`process(delivery)` ¶

Wraps the handling of one delivery; see the class docstring for the contract.

Parameters:

Name	Type	Description	Default
`delivery`	`Delivery`	The delivery being handled.	required

Returns:

Type	Description
`AbstractAsyncContextManager[None]`	An async context manager entered around the handling.

`start()` `async` ¶

Connects to the broker; called once before deliveries.

`stop()` `async` ¶

Disconnects from the broker; called once after the iterator is closed.

Bases: Protocol

A received message, settled exactly once by ack or reject.

A second ack or reject raises RuntimeError.

`body` `property` ¶

The message bytes, as given to Producer.publish.

`settled` `property` ¶

Whether ack or reject has completed.

`ack()` `async` ¶

Acknowledges the message so the broker does not redeliver it.

`reject(*, requeue=False)` `async` ¶

Rejects the message.

Parameters:

Name	Type	Description	Default
`requeue`	`bool`	If True, the broker redelivers the message.	`False`

Producer that publishes to a Redis stream, optionally after a delay.

`init(url=None, *, client=None, queue='default')` ¶

Initializes the producer.

Parameters:

Name	Type	Description	Default
`url`	`str \| None`	Redis URL, such as `redis://localhost:6379/0`. The producer creates its own client; `close` closes it. Give either `url` or `client`.	`None`
`client`	`Redis \| None`	Redis asyncio client, created with `decode_responses=False`. The caller owns it and closes it.	`None`
`queue`	`str`	Name of the queue to publish to.	`'default'`

Raises:

Type	Description
`ValueError`	If not exactly one of `url` and `client` is given.

`close()` `async` ¶

Closes the client if the producer created it.

`publish(envelope, body, options=None)` `async` ¶

Adds a message to the queue, held back for options.delay seconds if given.

Parameters:

Name	Type	Description	Default
`envelope`	`Envelope`	The envelope the message carries.	required
`body`	`bytes`	The encoded message to send.	required
`options`	`PublishOptions \| None`	The publish options.	`None`

Returns:

Type	Description
`str`	The stream entry id for an immediate message, or the envelope id
`str`	for a delayed one.

Consumer that reads a Redis stream through a consumer group.

process acks on success and rejects without requeue on any exception.

`connected` `property` ¶

Whether the last command reached Redis and the consumer is started.

`init(url=None, *, client=None, queue='default', group='asynk', lease=60.0, block=1.0, max_deliveries=None, dead_maxlen=10000)` ¶

Initializes the consumer.

Parameters:

Name	Type	Description	Default
`url`	`str \| None`	Redis URL, such as `redis://localhost:6379/0`. The consumer creates and closes its own client. Give either `url` or `client`.	`None`
`client`	`Redis \| None`	Redis asyncio client, created with `decode_responses=False`. The caller owns it and closes it.	`None`
`queue`	`str`	Name of the queue to consume.	`'default'`
`group`	`str`	Name of the consumer group; consumers in one group share the queue's messages.	`'asynk'`
`lease`	`float`	Seconds a delivered message may stay unsettled before another consumer in the group takes it over.	`60.0`
`block`	`float`	Seconds each read waits for a message; also how often delayed messages are moved to the stream and expired leases are taken over.	`1.0`
`max_deliveries`	`int \| None`	Deliveries after which a message that keeps being taken over moves to the dead-letter stream; None sets no limit.	`None`
`dead_maxlen`	`int`	Approximate number of entries the dead-letter stream keeps.	`10000`

Raises:

Type	Description
`ValueError`	If not exactly one of `url` and `client` is given, or `lease`, `block`, `max_deliveries` or `dead_maxlen` is not positive.

`deliveries()` `async` ¶

Yields a delivery for each message until stop.

`process(delivery)` `async` ¶

Acks on a clean exit and rejects without requeue on an exception.

Parameters:

Name	Type	Description	Default
`delivery`	`Delivery`	The delivery being handled.	required

Yields:

Type	Description
`AsyncIterator[None]`	None, while the delivery is handled.

`start()` `async` ¶

Connects to Redis and creates the consumer group if it does not exist.

A new group starts at the beginning of the stream, so messages published before it existed are consumed.

`stop()` `async` ¶

Ends deliveries and disconnects.

Messages still unsettled stay pending for the other consumers of the group to take over once their lease expires.

Codec¶

Bases: Protocol

Converts values to and from bytes.

`decode(data)` ¶

Deserializes a value.

Parameters:

Name	Type	Description	Default
`data`	`bytes`	Bytes previously produced by `encode`.	required

Returns:

Type	Description
`object`	The decoded value.

Raises:

Type	Description
`DecodeError`	If `data` is not a valid encoded value.

`encode(value)` ¶

Serializes a value.

Parameters:

Name	Type	Description	Default
`value`	`object`	The value to serialize.	required

Returns:

Type	Description
`bytes`	The encoded value.

Raises:

Type	Description
`EncodeError`	If this codec cannot encode `value`.

`validate(value)` ¶

Checks that encode would accept a value, without encoding it.

Parameters:

Name	Type	Description	Default
`value`	`object`	The value to check.	required

Raises:

Type	Description
`EncodeError`	If this codec cannot encode `value`.

Codec that encodes values as compact UTF-8 JSON.

Accepts None, bool, int, float, str, lists and str-keyed dicts. NaN and infinite floats are rejected in both directions.

`decode(data)` ¶

Deserializes JSON bytes into a value.

Parameters:

Name	Type	Description	Default
`data`	`bytes`	A UTF-8 encoded JSON document.	required

Returns:

Type	Description
`object`	The decoded value.

Raises:

Type	Description
`DecodeError`	If `data` is not a valid JSON document.

`encode(value)` ¶

Serializes a JSON-friendly value to JSON bytes.

Parameters:

Name	Type	Description	Default
`value`	`object`	The value to serialize.	required

Returns:

Type	Description
`bytes`	The UTF-8 encoded JSON document.

Raises:

Type	Description
`EncodeError`	If `value` is not JSON-friendly.

`validate(value)` ¶

Checks that a value is JSON-friendly.

Parameters:

Name	Type	Description	Default
`value`	`object`	The value to check.	required

Raises:

Type	Description
`EncodeError`	If `value` holds something JSON cannot represent, or a dict with non-str keys.

The message exchanged between producers and consumers.

Attributes:

Name	Type	Description
`task`	`str`	Registered name of the task to run.
`args`	`tuple[object, ...]`	Positional arguments for the task.
`kwargs`	`dict[str, object]`	Keyword arguments for the task.
`id`	`str`	Unique message id.
`attempt`	`int`	Zero-based delivery attempt counter.
`metadata`	`dict[str, object]`	Free-form data carried alongside the task call.

`create(task, args, kwargs, metadata=None)` `classmethod` ¶

Builds a new envelope with a fresh id.

Parameters:

Name	Type	Description	Default
`task`	`str`	Registered name of the task to run.	required
`args`	`Sequence[object]`	Positional arguments for the task.	required
`kwargs`	`Mapping[str, object]`	Keyword arguments for the task.	required
`metadata`	`Mapping[str, object] \| None`	Optional data carried alongside the call.	`None`

Returns:

Type	Description
`Envelope`	A new envelope at attempt 0.

`from_dict(data)` `classmethod` ¶

Builds an envelope from the mapping produced by to_dict.

Parameters:

Name	Type	Description	Default
`data`	`object`	A mapping with the keys `id`, `task`, `args`, `kwargs`, `attempt` and `metadata`.	required

Returns:

Type	Description
`Envelope`	The envelope the mapping describes.

Raises:

Type	Description
`InvalidEnvelopeError`	If `data` is not a mapping or a key is missing or has the wrong type.

`retry()` ¶

Returns a copy of this envelope with attempt incremented.

`to_dict()` ¶

Returns the envelope as a mapping of plain values.

Returns:

Type	Description
`dict[str, object]`	A dict with the keys `id`, `task`, `args` (a list), `kwargs`,
`dict[str, object]`	`attempt` and `metadata`, accepted by `from_dict`.

Store¶

Bases: Protocol

Key-value storage for task results and other small pieces of state.

Values are bytes.

`delete(key)` `async` ¶

Removes a key.

Parameters:

Name	Type	Description	Default
`key`	`str`	The key to remove.	required

Returns:

Type	Description
`bool`	Whether the key existed and was removed.

`get(key)` `async` ¶

Reads a value.

Parameters:

Name	Type	Description	Default
`key`	`str`	The key to read.	required

Returns:

Type	Description
`bytes \| None`	The stored value, or None if the key is missing.

`set(key, value, *, overwrite=False)` `async` ¶

Writes a value.

Parameters:

Name	Type	Description	Default
`key`	`str`	The key to write.	required
`value`	`bytes`	The value to store.	required
`overwrite`	`bool`	Whether to replace an existing value.	`False`

Returns:

Type	Description
`bool`	True if the value was written, False if the key already existed
`bool`	and `overwrite` was false.

`start()` `async` ¶

Acquires the resources the store needs.

`stop()` `async` ¶

Releases the resources acquired by start.

`wait(key, timeout=None)` `async` ¶

Waits until a key has a value.

Parameters:

Name	Type	Description	Default
`key`	`str`	The key to wait for.	required
`timeout`	`float \| None`	Seconds to wait.	`None`

Returns:

Type	Description
`bytes \| None`	The value, or None if `timeout` elapsed first.

Store that keeps values in memory within one process.

`init(ttl=86400.0, *, clock=time.monotonic)` ¶

Initializes the store.

Parameters:

Name	Type	Description	Default
`ttl`	`float`	Seconds a key lives after it is written.	`86400.0`
`clock`	`Callable[[], float]`	Monotonic time source, replaceable for tests.	`monotonic`

`delete(key)` `async` ¶

Removes a key; returns False if it is missing or expired.

`get(key)` `async` ¶

Reads a value, or returns None if the key is missing or expired.

`set(key, value, *, overwrite=False)` `async` ¶

Writes a value; returns False if the key exists and overwrite is false.

`start()` `async` ¶

Does nothing.

`stop()` `async` ¶

Does nothing.

`wait(key, timeout=None)` `async` ¶

Waits until the key has a value; returns None if timeout elapses first.

Store for apps with no store configured.

set always reports a write, delete reports that nothing was removed, and get and wait raise NotConfiguredError.

`delete(key)` `async` ¶

Returns False: nothing is stored.

`get(key)` `async` ¶

Raises NotConfiguredError.

`set(key, value, *, overwrite=False)` `async` ¶

Discards the value and returns True.

`start()` `async` ¶

Does nothing.

`stop()` `async` ¶

Does nothing.

`wait(key, timeout=None)` `async` ¶

Raises NotConfiguredError.

Store on Redis that several processes and machines share.

`init(url=None, *, client=None, ttl=86400.0, prefix='asynk:', max_poll_interval=1.0)` ¶

Initializes the store.

Parameters:

Name	Type	Description	Default
`url`	`str \| None`	Redis URL, such as `redis://localhost:6379/0`. The store creates and closes its own client. Give either `url` or `client`.	`None`
`client`	`Redis \| None`	Redis asyncio client, created with `decode_responses=False`. The caller owns it and closes it.	`None`
`ttl`	`float`	Seconds a key lives after it is written.	`86400.0`
`prefix`	`str`	Text put before every key, so several apps can share one Redis.	`'asynk:'`
`max_poll_interval`	`float`	Longest pause in seconds between the checks `wait` makes for a value.	`1.0`

Raises:

Type	Description
`ValueError`	If not exactly one of `url` and `client` is given, or `ttl` or `max_poll_interval` is not positive.

`delete(key)` `async` ¶

Removes a key; returns False if it is missing or expired.

`get(key)` `async` ¶

Reads a value, or returns None if the key is missing or expired.

`set(key, value, *, overwrite=False)` `async` ¶

Writes a value; returns False if the key exists and overwrite is false.

`start()` `async` ¶

Does nothing: the client connects on first use.

`stop()` `async` ¶

Closes the client if the store created it.

`wait(key, timeout=None)` `async` ¶

Polls until the key has a value; returns None if timeout elapses first.

Parameters:

Name	Type	Description	Default
`key`	`str`	Key to wait for.	required
`timeout`	`float \| None`	Seconds to wait, or None to wait until the key has a value.	`None`

Calls fetch until it returns a value, pausing longer after each miss.

Parameters:

Name	Type	Description	Default
`fetch`	`Callable[[], Awaitable[bytes \| None]]`	Coroutine function that returns a value, or None if there is none yet.	required
`timeout`	`float \| None`	Seconds to keep trying, or None to try until there is a value.	required
`max_interval`	`float`	Longest pause in seconds between two calls.	required

Returns:

Type	Description
`bytes \| None`	The value, or None if `timeout` elapsed first.

Telemetry¶

Bases: Protocol

Tracing and metrics hooks an app calls at fixed points.

An implementation must not raise: telemetry never breaks publishing or tasks.

`layer` `property` ¶

Middleware a worker runs outside every layer of the app, or None.

`inject(metadata)` ¶

Adds trace context to the metadata of an envelope about to be published.

Keys already in metadata are kept.

Parameters:

Name	Type	Description	Default
`metadata`	`dict[str, object]`	The metadata of the envelope; changed in place.	required

`message_published(envelope)` ¶

Records that an envelope was published.

Parameters:

Name	Type	Description	Default
`envelope`	`Envelope`	The envelope that was sent.	required

`retry_published(envelope)` ¶

Records that a worker published a retry.

Parameters:

Name	Type	Description	Default
`envelope`	`Envelope`	The envelope of the attempt that is being retried.	required

`task_finished()` ¶

Records that a worker finished handling a delivery, however it ended.

`task_started()` ¶

Records that a worker began handling a delivery.

Telemetry that does nothing; the one an app uses when OpenTelemetry is off.

Attributes:

Name	Type	Description
`layer`	`Middleware \| None`	Always None.

`inject(metadata)` ¶

Does nothing.

`message_published(envelope)` ¶

Does nothing.

`retry_published(envelope)` ¶

Does nothing.

`task_finished()` ¶

Does nothing.

`task_started()` ¶

Does nothing.

Traces and measures an app's publishing and tasks with OpenTelemetry.

Publishing writes the active trace context into the envelope metadata. A worker runs each task under a CONSUMER span whose parent is that context, and records the metrics messaging.process.duration, messaging.client.sent.messages, messaging.client.consumed.messages, asynk.worker.active_tasks and asynk.task.retries.

Any error in this class is logged once and turns all of its telemetry off for the process; tasks and publishing carry on.

Attributes:

Name	Type	Description
`layer`		The middleware a worker runs outside every layer of the app.

`init(*, tracer_provider=None, meter_provider=None)` ¶

Initializes the component.

Parameters:

Name	Type	Description	Default
`tracer_provider`	`TracerProvider \| None`	Provider of the tracer; defaults to the global one.	`None`
`meter_provider`	`MeterProvider \| None`	Provider of the meter; defaults to the global one.	`None`

`inject(metadata)` ¶

Adds the active trace context to the metadata; keys already there are kept.

Parameters:

Name	Type	Description	Default
`metadata`	`dict[str, object]`	The metadata of the envelope; changed in place.	required

`message_published(envelope)` ¶

Counts a published message.

Parameters:

Name	Type	Description	Default
`envelope`	`Envelope`	The envelope that was sent.	required

`retry_published(envelope)` ¶

Counts a published retry.

Parameters:

Name	Type	Description	Default
`envelope`	`Envelope`	The envelope of the attempt that is being retried.	required

`task_finished()` ¶

Stops counting a delivery as active.

`task_started()` ¶

Counts a delivery as active.

Results¶

Bases: Generic[R]

Reference to a published task, used to read its state and result.

R is the type the task's function returns. It is a declaration for type checkers, not checked at runtime.

Attributes:

Name	Type	Description
`task_id`		The task's id (`Envelope.id`).
`receipt`		Whatever the producer returned when publishing, if anything.

`init(task_id, store, codec, receipt=None)` ¶

Initializes the handle.

Parameters:

Name	Type	Description	Default
`task_id`	`str`	The task's id (`Envelope.id`).	required
`store`	`Store`	The store holding the task's records.	required
`codec`	`Codec`	The codec the records were encoded with.	required
`receipt`	`Any`	Whatever the producer returned when publishing, if anything.	`None`

`result(timeout=None)` `async` ¶

Waits for the task to finish and returns its value.

Parameters:

Name	Type	Description	Default
`timeout`	`float \| None`	Seconds to wait.	`None`

Returns:

Type	Description
`R \| None`	The task's return value, or None if it was skipped.

Raises:

Type	Description
`NotConfiguredError`	If no store is configured.
`TaskNotFoundError`	If the store has no record of the task.
`TimeoutError`	If no result arrived within `timeout`.
`TaskError`	If the task failed.

`state()` `async` ¶

Reads the task's state; UNKNOWN if the store has no record of it.

Raises:

Type	Description
`NotConfiguredError`	If no store is configured.

Bases: Enum

Lifecycle state of a task as recorded in the store.

Attributes:

Name	Type	Description
`PENDING`		Published, not started yet.
`RUNNING`		A worker is running it.
`RETRYING`		An attempt failed with retries left and will be retried.
`SUCCESS`		Finished; `TaskHandle.result` returns the value.
`FAILURE`		Failed for good; `TaskHandle.result` raises `TaskError`.
`SKIPPED`		Middleware skipped the task; `TaskHandle.result` returns None.
`UNKNOWN`		The store has no record for this id.

Result of handling one delivery.

Attributes:

Name	Type	Description
`envelope`	`Envelope`	The envelope that was handled.
`status`	`TaskStatus`	How the run ended.
`value`	`Any`	The task's return value on success.
`error`	`BaseException \| None`	The raised exception on failure.

Bases: Enum

Outcome of running a task, as seen by middleware and the consumer.

Attributes:

Name	Type	Description
`SUCCESS`		The task returned normally.
`FAILED`		The task, or a layer before it ran, raised and is not retried: no retry policy applies, the exception is not in its `retry_on`, or the layer settled the delivery.
`RETRYING`		The task, or a layer before it ran, raised and the retry policy has retries left.
`RETRIES_EXCEEDED`		The task, or a layer before it ran, raised and the retry policy has no retries left.
`UNKNOWN_TASK`		The envelope names a task this app has not registered.
`SKIPPED`		Middleware skipped the task, so it did not run.

Errors¶

Bases: Exception

Base class of every exception asynk defines.

Bases: AsynkError

Raised when an app lacks a component required by the operation.

Bases: AsynkError

Raised when a value holds something the codec cannot encode.

Bases: AsynkError

Raised when bytes cannot be decoded into a valid value.

Bases: AsynkError

Raised when data is not a valid envelope.

Bases: AsynkError

Raised when a cron expression is invalid.

Bases: AsynkError

Raised by TaskHandle.result when the task failed.

Attributes:

Name	Type	Description
`type_name`		Name of the exception type the task raised.
`message`		The exception's message.

Bases: AsynkError

Raised by TaskHandle.result for a task id the store does not know.

Bases: AsynkError

Raised inside a worker when a task attempt exceeds its timeout.

Bases: AsynkError

Raised in Consumer.process for a failed, unknown or retries-exceeded task.

Attributes:

Name	Type	Description
`result`		The failed task's result.

API reference¶

App and tasks¶

scheduled_tasks property ¶

tasks property ¶

__init__(name, *, producer=None, consumer=None, store=None, codec=None, tz='UTC', options=None) ¶

after_shutdown(hook) ¶

after_startup(hook) ¶

before_shutdown(hook) ¶

before_startup(hook) ¶

get_task(task_id) ¶

middleware(layer) ¶

publish(envelope, options=None) async ¶

registered_task(name) ¶

scheduled_task(*, cron, tz=None, args=(), kwargs=None, name=None, options=None) ¶

send(envelope, options=None) async ¶

task(*, name=None, options=None) ¶

__call__(*args, **kwargs) ¶

delay(*args, **kwargs) async ¶

with_metadata(**metadata) ¶

policy property ¶

with_defaults(defaults) ¶

__init__(app, *, clock=None, probe_dir=None, heartbeat_interval=5.0, grace=30.0) ¶

request_shutdown() ¶

run() async ¶

run_with_signals() async ¶

Running¶

__init__(app, *, concurrency=10, shutdown_timeout=10.0, probe_dir=None, heartbeat_interval=5.0) ¶

handle(delivery) async ¶

request_shutdown() ¶

run() async ¶

run_with_signals() async ¶

__init__(app, *, probe_dir=None, heartbeat_interval=5.0) ¶

request_shutdown() ¶

run() async ¶

run_with_signals() async ¶

Middleware¶

skip() ¶

Retries¶

delay(retry) ¶

delay(retry) ¶

Transport protocols¶

publish(envelope, body, options=None) async ¶

__post_init__() ¶

connected property ¶

deliveries() ¶

process(delivery) ¶

start() async ¶

stop() async ¶

body property ¶

settled property ¶

ack() async ¶

reject(*, requeue=False) async ¶

__init__(url=None, *, client=None, queue='default') ¶

close() async ¶

publish(envelope, body, options=None) async ¶

connected property ¶

__init__(url=None, *, client=None, queue='default', group='asynk', lease=60.0, block=1.0, max_deliveries=None, dead_maxlen=10000) ¶

deliveries() async ¶

process(delivery) async ¶

start() async ¶

stop() async ¶

Codec¶

decode(data) ¶

encode(value) ¶

validate(value) ¶

decode(data) ¶

encode(value) ¶

validate(value) ¶

create(task, args, kwargs, metadata=None) classmethod ¶

from_dict(data) classmethod ¶

retry() ¶

to_dict() ¶

Store¶

delete(key) async ¶

get(key) async ¶

set(key, value, *, overwrite=False) async ¶

start() async ¶

stop() async ¶

wait(key, timeout=None) async ¶

__init__(ttl=86400.0, *, clock=time.monotonic) ¶

`scheduled_tasks` `property` ¶

`tasks` `property` ¶

`init(name, *, producer=None, consumer=None, store=None, codec=None, tz='UTC', options=None)` ¶

`after_shutdown(hook)` ¶

`after_startup(hook)` ¶

`before_shutdown(hook)` ¶

`before_startup(hook)` ¶

`get_task(task_id)` ¶

`middleware(layer)` ¶

`publish(envelope, options=None)` `async` ¶

`registered_task(name)` ¶

`scheduled_task(*, cron, tz=None, args=(), kwargs=None, name=None, options=None)` ¶

`send(envelope, options=None)` `async` ¶

`task(*, name=None, options=None)` ¶

`call(*args, **kwargs)` ¶

`delay(*args, **kwargs)` `async` ¶

`with_metadata(**metadata)` ¶

`policy` `property` ¶

`with_defaults(defaults)` ¶

`init(app, *, clock=None, probe_dir=None, heartbeat_interval=5.0, grace=30.0)` ¶

`request_shutdown()` ¶

`run()` `async` ¶

`run_with_signals()` `async` ¶

`init(app, *, concurrency=10, shutdown_timeout=10.0, probe_dir=None, heartbeat_interval=5.0)` ¶

`handle(delivery)` `async` ¶

`request_shutdown()` ¶

`run()` `async` ¶

`run_with_signals()` `async` ¶

`init(app, *, probe_dir=None, heartbeat_interval=5.0)` ¶

`request_shutdown()` ¶

`run()` `async` ¶

`run_with_signals()` `async` ¶

`skip()` ¶

`delay(retry)` ¶

`delay(retry)` ¶

`publish(envelope, body, options=None)` `async` ¶

`__post_init__()` ¶

`connected` `property` ¶

`deliveries()` ¶

`process(delivery)` ¶

`start()` `async` ¶

`stop()` `async` ¶

`body` `property` ¶

`settled` `property` ¶

`ack()` `async` ¶

`reject(*, requeue=False)` `async` ¶

`init(url=None, *, client=None, queue='default')` ¶

`close()` `async` ¶

`publish(envelope, body, options=None)` `async` ¶

`connected` `property` ¶

`init(url=None, *, client=None, queue='default', group='asynk', lease=60.0, block=1.0, max_deliveries=None, dead_maxlen=10000)` ¶

`deliveries()` `async` ¶

`process(delivery)` `async` ¶

`start()` `async` ¶

`stop()` `async` ¶

`decode(data)` ¶

`encode(value)` ¶

`validate(value)` ¶

`decode(data)` ¶

`encode(value)` ¶

`validate(value)` ¶

`create(task, args, kwargs, metadata=None)` `classmethod` ¶

`from_dict(data)` `classmethod` ¶

`retry()` ¶

`to_dict()` ¶

`delete(key)` `async` ¶

`get(key)` `async` ¶

`set(key, value, *, overwrite=False)` `async` ¶

`start()` `async` ¶

`stop()` `async` ¶

`wait(key, timeout=None)` `async` ¶

`init(ttl=86400.0, *, clock=time.monotonic)` ¶

`delete(key)` `async` ¶

`get(key)` `async` ¶

`set(key, value, *, overwrite=False)` `async` ¶

`start()` `async` ¶

`stop()` `async` ¶

`wait(key, timeout=None)` `async` ¶

`delete(key)` `async` ¶

`get(key)` `async` ¶