Writing a custom transport¶

A transport connects asynk to a broker. It is two objects that satisfy the Producer and Consumer protocols, plus the Delivery objects the consumer yields. They are protocols, so no base class is needed: any objects with the right methods work, and mypy checks them where they are passed to App.

This one runs over an asyncio.Queue. A broker transport has the same shape, with network calls in place of the queue:

"""A minimal transport: a producer, a consumer and their deliveries over an asyncio queue.

A real transport talks to a broker in the same places: `publish` sends the body,
`deliveries` receives messages, and `ack` and `reject` settle them.
"""

import asyncio
from collections.abc import AsyncIterator
from contextlib import asynccontextmanager

from asynk import App, Delivery, Envelope, MemoryStore, PublishOptions, Worker

Queue = asyncio.Queue[bytes | None]


class QueueProducer:
    """Sends message bodies to the queue."""

    def __init__(self, queue: Queue) -> None:
        self._queue = queue

    async def publish(
        self, envelope: Envelope, body: bytes, options: PublishOptions | None = None
    ) -> str:
        # The queue cannot delay a message, so `options` is ignored.
        await self._queue.put(body)
        return envelope.id  # the receipt, available as handle.receipt


class QueueDelivery:
    """One received message, settled once."""

    def __init__(self, body: bytes, queue: Queue) -> None:
        self._body = body
        self._queue = queue
        self._settled = False

    @property
    def body(self) -> bytes:
        return self._body

    @property
    def settled(self) -> bool:
        return self._settled

    async def ack(self) -> None:
        self._settled = True

    async def reject(self, *, requeue: bool = False) -> None:
        self._settled = True
        if requeue:
            await self._queue.put(self._body)


class QueueConsumer:
    """Receives messages from the queue and settles each one."""

    def __init__(self, queue: Queue) -> None:
        self._queue = queue
        self._connected = False

    @property
    def connected(self) -> bool:
        return self._connected

    async def start(self) -> None:
        self._connected = True

    async def stop(self) -> None:
        self._connected = False
        self._queue.put_nowait(None)

    async def deliveries(self) -> AsyncIterator[Delivery]:
        while (body := await self._queue.get()) is not None:
            yield QueueDelivery(body, self._queue)

    @asynccontextmanager
    async def process(self, delivery: Delivery) -> AsyncIterator[None]:
        try:
            yield
        except BaseException:
            if not delivery.settled:
                await delivery.reject()
            raise  # never suppress, cancellation included
        if not delivery.settled:
            await delivery.ack()


queue: Queue = asyncio.Queue()
app = App(
    "custom-transport",
    producer=QueueProducer(queue),
    consumer=QueueConsumer(queue),
    store=MemoryStore(),
)


@app.task()
async def add(a: int, b: int) -> int:
    return a + b


async def main() -> None:
    worker = Worker(app)
    run = asyncio.create_task(worker.run())

    handle = await add.delay(2, b=3)
    print("receipt:", handle.receipt == handle.task_id)
    print("2 + 3 =", await handle.result(timeout=5))

    worker.request_shutdown()
    await run


if __name__ == "__main__":
    asyncio.run(main())

The signatures and their docstrings are in the API reference: Producer, Consumer and Delivery, with PublishOptions.

Producer¶

publish(envelope, body, options=None) sends body, the bytes the app built with app.codec.encode(envelope.to_dict()). The envelope is there for routing and broker metadata, such as using envelope.id as the broker's message id. The transport never encodes anything itself. Whatever publish returns becomes handle.receipt.

options is a PublishOptions or None. Its delay asks the broker to hold the message for that many seconds before delivering it; the worker passes it when it publishes a retry with backoff, and only then, so a producer that takes only (envelope, body) still works for retries that have no backoff. Options are advisory: honor what your broker supports and ignore the rest. A transport that ignores delay delivers retries immediately, so they get no backoff. Raising from publish instead is worse: the worker then rejects the original delivery with requeue, so it runs again at the same attempt and never uses up its retries. Broker limits, such as a maximum delay, are the transport's to handle. The memory transport holds a delayed message in a timer, so one not yet due is lost when the process exits.

Delivery¶

A delivery carries the received body and is settled exactly once, with ack() or reject(requeue=...); a second call raises RuntimeError. settled reports whether that happened. The worker settles some deliveries itself (see below), so process checks settled before settling.

The consumer contract¶

start() is called once before deliveries(), and stop() once after the iterator is closed. Deliveries still unsettled when stop() runs should be redelivered by the broker.
connected drives the readiness probe: report false while the broker connection is down.
deliveries() returns the stream of incoming deliveries. It may be cancelled at any await point during shutdown; if the iterator has an aclose() method the worker calls it. When the iterator ends, the worker drains its running tasks and Worker.run() returns.
process(delivery) is an async context manager entered once per delivery, around its handling. On a clean exit it must settle an unsettled delivery (usually ack). On an exception it must settle an unsettled delivery (usually reject) and must never suppress the exception, especially asyncio.CancelledError.

What reaches process:

When a task fails for good, is unknown or exceeds its retries, the worker raises DeliveryFailedError inside the block; .result.status says which. A task that is retried is settled by the worker: it publishes the retry and acks the delivery, or rejects it with requeue when the retry cannot be published. Any other exception is an unexpected error.
The worker decodes each body inside process. A body it cannot decode is rejected without requeue there, and the block exits cleanly.
When a task is cancelled at the shutdown timeout, the worker rejects its unsettled delivery with requeue=True before the cancellation reaches process, so the message is redelivered rather than dropped.

Testing your transport¶

The asynk repository holds a contract test suite that checks the behavior described above against your transport. To use it:

Copy tests/transport/contract/ from the asynk repository into your tests.
Replace REGISTRY in conftest.py with a factory for your transport. The factory returns a TransportHandle: a producer, a consumer, a way to make more consumers on the same queue, and a close() that stops them and deletes the queue. Give each handle a fresh queue name.
Install pytest-asyncio; the suite is written as async tests.

The header of conftest.py names the asynk version the suite matches. Changes to the contract appear in the generated changelog.

The suite checks delayed publish when supports_delay is true in your ContractSettings (the default); set it to false if your transport ignores PublishOptions.delay. It does not yet cover message ordering.

The asynk repository runs the suite against the Redis transport twice: on fakeredis and on a real Redis server. The real-server entry is always collected. It is skipped when ASYNK_TEST_REDIS_URL is unset or no server answers there, and it fails instead when ASYNK_TEST_REQUIRE_SERVICES=1, which CI sets. To run it locally with Docker:

docker run --rm -d --name asynk-redis -p 6379:6379 redis:7.4.12
ASYNK_TEST_REDIS_URL=redis://localhost:6379/0 ASYNK_TEST_REQUIRE_SERVICES=1 \
  uv run pytest tests/transport
docker stop asynk-redis

The image is the one CI uses. What the fake cannot prove is listed under Testing in the Redis transport guide.