Modules

An overview of the packages that make up litestar_events. The API reference below is generated automatically from the source docstrings.

Package (litestar_events)

Shared emitter machinery, configuration, and the Litestar plugin.

Top-level package for the project.

Configuration shared by the producer and the worker process.

class litestar_events.config.EventsConfig(backend=None, backend_options=<factory>, role=None, codec=None, typed_decoding=True, local_listeners=(), request_id_header=None, request_id_exclude_path_pattern=None, echo_request_id=True, drain_timeout=30.0)

Bases: object

Settings for EventsPlugin.

Import the same config in the web app and the worker: the worker loads the same Litestar app (litestar events run) and only the role differs.

Parameters:

• backend (type[BaseEventEmitterBackend] | None)

• backend_options (dict[str, Any])

• role (EventsRole | str | None)

• codec (Codec | None)

• typed_decoding (bool)

• local_listeners (Sequence[EventListener])

• request_id_header (str | None)

• request_id_exclude_path_pattern (str | tuple[str, ...] | None)

• echo_request_id (bool)

• drain_timeout (float)

backend: type[BaseEventEmitterBackend] | None = None

Emitter backend to install on the app. None keeps the app’s own.

backend_options: dict[str, Any]

Keyword arguments forwarded to the backend constructor.

role: EventsRole | str | None = None

Role for this process; events run always uses worker.

codec: Codec | None = None

Wire format shared by producer and worker; default is JSONCodec.

typed_decoding: bool = True

Convert worker-side arguments to the listener’s annotated types.

Only applies to the default codec; a custom codec decides for itself.

local_listeners: Sequence[EventListener] = ()

Listeners that run in the web (producer) process instead of the worker.

A producer calls them directly from app.emit() without the broker; a worker skips them. Each must also be in the app’s listeners.

request_id_header: str | None = None

Request header whose value becomes the correlation id of events emitted while serving the request (e.g. "x-request-id"). None adds no middleware.

request_id_exclude_path_pattern: str | tuple[str, ...] | None = None

Regex (or regexes) of paths the request-id middleware skips, such as "^/health". Routes can also opt out with opt={"exclude_request_id": True}.

echo_request_id: bool = True

Also set the request id on the response (needs request_id_header).

drain_timeout: float = 30.0

Default seconds events drain waits for in-flight listeners.

class litestar_events.config.EventsRole(*values)

Bases: str, Enum

What an emitter does in the current process.

EMBEDDED = 'embedded'

Publish and consume in the same process (the historical behavior).

PRODUCER = 'producer'

no broker subscription, no consumer task.

Type:

Publish only

WORKER = 'worker'

listeners run here, emit is rejected.

Type:

Consume only

litestar_events.config.resolve_role(role=None)

Resolve role to an EventsRole; None means embedded.

Raises:

ValueError – if role is not a valid role.

Parameters:

role (EventsRole | str | None)

Return type:

EventsRole

Litestar plugin wiring the emitter backend and the events CLI group.

class litestar_events.plugin.EventsPlugin(config=None)

Bases: InitPluginProtocol, CLIPlugin

Install an emitter backend and add the litestar events commands.

Use the same plugin and config in the web app and the worker. The worker runs the same Litestar app through litestar events run, which always consumes only; set EventsConfig(role="producer") so the web process publishes only.

Plugin order matters for the request id: with request_id_header set, the middleware goes first in the app’s list when this plugin initializes, so list EventsPlugin last in plugins=[...] to keep it outermost.

Parameters:

config (EventsConfig | None)

on_app_init(app_config)

Receive the AppConfig instance after on_app_init hooks have been called.

Examples

from litestar import Litestar, get
from litestar.di import Provide
from litestar.plugins import InitPluginProtocol


def get_name() -> str:
    return "world"


@get("/my-path")
def my_route_handler(name: str) -> dict[str, str]:
    return {"hello": name}


class MyPlugin(InitPluginProtocol):
    def on_app_init(self, app_config: AppConfig) -> AppConfig:
        app_config.dependencies["name"] = Provide(get_name)
        app_config.route_handlers.append(my_route_handler)
        return app_config


app = Litestar(plugins=[MyPlugin()])

Parameters:

• app_config – The AppConfig instance.

• app_config (AppConfig)

Returns:

The app config object.

Return type:

AppConfig

on_cli_init(cli)

Called when the CLI is initialized.

This can be used to extend or override existing commands.

Parameters:

• cli – The root click.Group of the Litestar CLI

• cli (Group)

Return type:

None

Examples

from litestar import Litestar
from litestar.plugins import CLIPluginProtocol
from click import Group


class CLIPlugin(CLIPluginProtocol):
    def on_cli_init(self, cli: Group) -> None:
        @cli.command()
        def is_debug_mode(app: Litestar):
            print(app.debug)


app = Litestar(plugins=[CLIPlugin()])

Wire format and listener-argument binding shared by every backend.

The producer and the worker must agree on the bytes that travel through the broker. This module is that single definition: JSONCodec encodes emit(event_id, *args, **kwargs) into a versioned JSON envelope and decodes it again on the worker, optionally converting the decoded values into the types annotated on the listener’s parameters.

litestar_events.codec.WIRE_VERSION = 1

Version stamped on every message ("v"). Messages without it are version 1 messages written before the field existed. The envelope fields added since ("cid", "h") are optional, so the version is unchanged.

class litestar_events.codec.Codec(*args, **kwargs)

Bases: Protocol

How events cross the broker and reach listeners.

encode(message)

Serialize an event. Called synchronously from emit.

Parameters:

message (Message)

Return type:

bytes

decode(data)

Parse a message back into a Message.

Parameters:

data (bytes | str)

Return type:

Message

invoke(fn, event_id, message)

Call a listener with a decoded message.

Parameters:

• fn (Callable[..., Awaitable[Any]])

• event_id (str)

• message (Message)

Return type:

Awaitable[Any]

class litestar_events.codec.JSONCodec(*, typed=True, type_encoders=None, type_decoders=None)

Bases: object

JSON envelope: v, cid, h, args and kwargs keys.

Encoding uses Litestar’s serializer, so datetime, UUID, Decimal, dataclasses and msgspec/Pydantic models are all supported on the producer.

Parameters:

• typed – when True (default), the worker converts each decoded argument to the type annotated on the listener’s parameter (order: Order, when: datetime) before calling it. Unannotated parameters, and listeners whose annotations cannot be resolved, get plain JSON values. When False, every listener gets plain JSON values, as in earlier releases.

• type_encoders – extra Litestar-style type encoders for encode.

• type_decoders – extra (predicate, decoder) pairs for typed decoding.

• typed (bool)

• type_encoders (dict[Any, Callable[[Any], Any]] | None)

• type_decoders (Sequence[tuple[Callable[[Any], bool], Callable[[Any, Any], Any]]] | None)

A listener whose single parameter is a model (dataclass, msgspec Struct, Pydantic model or TypedDict) can receive the whole keyword payload: app.emit("order_created", id=1, total=9) populates order: Order. A parameter annotated EventContext is injected with the event’s id, headers and correlation id.

content_type = 'application/json'

Advertised on brokers with a native content-type field (RabbitMQ).

class litestar_events.codec.Message(args=(), kwargs=<factory>, headers=<factory>, correlation_id=None)

Bases: object

One event as it crosses the broker.

Parameters:

• args (tuple[Any, ...])

• kwargs (dict[str, Any])

• headers (dict[str, str])

• correlation_id (str | None)

litestar_events.codec.listener_name(fn)

Name of the user’s function behind a listener’s fn, for messages.

Parameters:

fn (Callable[..., Any])

Return type:

str

Event headers, correlation ids and the listener-side event context.

Every event carries a correlation id and optional string headers across the broker. Set them for the events emitted in a block with event_headers(); read them in a listener by annotating a parameter as EventContext or by calling current_event(). Events emitted from inside a listener inherit its correlation id, so one request can be traced across processes.

class litestar_events.context.EventContext(event_id, headers, correlation_id)

Bases: object

What a listener can know about the event it is handling.

Annotate a listener parameter with this type and it is injected:

@listener("order_created")
async def on_order(order_id: int, ctx: EventContext) -> None: ...

Parameters:

• event_id (str)

• headers (Mapping[str, str])

• correlation_id (str | None)

litestar_events.context.current_event()

Return the event the running listener is handling, else None.

Return type:

EventContext | None

litestar_events.context.event_headers(headers=None, /, *, correlation_id=None, **extra)

Attach headers (and optionally a correlation id) to events emitted inside.

Nested blocks merge, inner values winning. Header names and values are strings. Example:

with event_headers(tenant="acme", correlation_id=request_id):
    app.emit("order_created", order_id=1)

Raises:

TypeError – if a header name or value is not a string.

Parameters:

• headers (Mapping[str, str] | None)

• correlation_id (str | None)

• extra (str)

Return type:

Generator[None, None, None]

litestar_events.context.new_correlation_id()

Return a fresh correlation id.

Return type:

str

litestar_events.context.outgoing_correlation_id()

Correlation id for an event emitted now.

The explicit one from event_headers(), else the one of the event the current listener handles, else a new id.

Return type:

str

litestar_events.context.outgoing_headers()

Headers to stamp on an event emitted now.

Return type:

dict[str, str]

litestar_events.context.use_event(context)

Make context the current event; used while a listener runs.

Parameters:

context (EventContext)

Return type:

Generator[None, None, None]

ASGI middleware tying a request id to the events emitted while serving it.

class litestar_events.middleware.RequestIdMiddleware(*, header='x-request-id', echo=True, exclude_path_pattern=None)

Bases: ASGIMiddleware

Use the request’s id as the correlation id of events emitted during it.

The id is read from header (x-request-id by default). A missing or unsafe value is replaced with a generated one. Events emitted while the request is handled carry the id, so a worker’s logs can be tied back to the request. With echo the id is also set on the response.

Routes can opt out with opt={"exclude_request_id": True} or by path through exclude_path_pattern.

Add it through EventsConfig(request_id_header=...), or register it yourself as ordinary Litestar middleware.

Parameters:

• header (str)

• echo (bool)

• exclude_path_pattern (str | tuple[str, ...] | None)

async handle(scope, receive, send, next_app)

Handle ASGI call.

Parameters:

• scope – The ASGI connection scope.

• receive – The ASGI receive function.

• send – The ASGI send function

• next_app – The next ASGI application in the middleware stack to call

• scope (Scope)

• receive (Receive)

• send (Send)

• next_app (ASGIApp)

Return type:

None

Backends (litestar_events.contrib)

Drop-in BaseEventEmitterBackend implementations, one subpackage per transport. See each backend’s docstring for its delivery semantics and constructor arguments.