Publishing with metadata¶

Metadata is free-form data carried alongside a call, in envelope.metadata, for example a trace id or the source of the request. Attach it with with_metadata:

"""Publishing a task call with metadata attached to its envelope."""

import asyncio

from asynk import App, Envelope, PublishOptions


class PrintingProducer:
    """A producer that prints each envelope's metadata instead of sending it."""

    async def publish(
        self, envelope: Envelope, body: bytes, options: PublishOptions | None = None
    ) -> str:
        print(envelope.task, envelope.metadata)
        return envelope.id


app = App("metadata", producer=PrintingProducer())


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


async def main() -> None:
    await add.with_metadata(source="api").delay(1, b=2)

    # Each call returns a new bound task; later keys override earlier ones.
    traced = add.with_metadata(source="api", trace="x")
    await traced.with_metadata(trace="y").delay(1, b=2)


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

with_metadata returns an immutable bound task, so calls can be chained and later keys override earlier ones; the task itself is unchanged. The bound task has the same type-checked delay as the task. Metadata values go through the app's codec with the rest of the envelope, so they must be values it can encode.

The scheduler uses metadata too: each scheduled publish carries idempotency_key and scheduled_at.