litestar_events

Event Emitter Backends for Litestar.

Table of Contents

• Installation
• Backends
• Quick start
• Examples
• Delivery semantics
• Support
• Motivation
• About
• Author
• Analysis
• Contributing
• Development
• Releasing
• Credits
• License
• Changelog

Installation

litestar_events is a library. Add it to your project as a dependency:

uv add litestar_events

Or with pip:

pip install litestar_events

Install the core package plus the extra for the backend(s) you want:

# RabbitMQ backend
pip install "litestar_events[rabbit]"

# Multiple backends at once
pip install "litestar_events[rabbit,redis,nats]"

Using uv:

uv add "litestar_events[rabbit]"

Verify the installation

The package root is importable after installation:

>>> from litestar_events import __doc__
>>> isinstance(__doc__, str)
True

Backends

Quick start

The same Litestar app, three different transports. The app logic — your route handler, your listeners, your emitted events — is identical. Only the event_emitter_backend= line changes.

1. Define your event and listeners

# app.py
from __future__ import annotations

from typing import Any

from litestar import Request, post
from litestar.events import listener

USER_REGISTERED = "user_registered"


@listener(USER_REGISTERED)
async def send_welcome_email(*, email: str, **_: Any) -> None:
    print(f"[email] welcome to {email}")


@listener(USER_REGISTERED)
async def record_analytics(*, email: str, **_: Any) -> None:
    print(f"[analytics] new signup: {email}")


@post("/users")
async def register_user(request: Request, data: dict[str, str]) -> dict[str, str]:
    email = data["email"]
    request.app.emit(USER_REGISTERED, email=email)
    return {"status": "queued", "email": email}

Two listeners react concurrently to one event. Notice the listeners use **_: Any — that's the Litestar convention for keeping listener signatures forwards-compatible when multiple listeners share an event payload.

2. Pick a backend

Every backend constructor accepts listeners: Sequence[EventListener] plus its own keyword arguments. Use functools.partial to bake in the kwargs and hand the result to Litestar.

RabbitMQ (durable, at-least-once)

from functools import partial

from litestar import Litestar

from litestar_events.contrib.rabbit import RabbitEventEmitter

app = Litestar(
    route_handlers=[register_user],
    listeners=[send_welcome_email, record_analytics],
    event_emitter_backend=partial(
        RabbitEventEmitter,
        amqp_url="amqp://guest:guest@localhost/",
        # queue_name="my-app",   # set for work-queue semantics across replicas
    ),
)

Redis (fire-and-forget pub/sub)

from functools import partial

from litestar import Litestar

from litestar_events.contrib.redis import RedisEventEmitter

app = Litestar(
    route_handlers=[register_user],
    listeners=[send_welcome_email, record_analytics],
    event_emitter_backend=partial(
        RedisEventEmitter, redis_url="redis://localhost:6379/0"
    ),
)

PostgreSQL (LISTEN/NOTIFY)

from functools import partial

from litestar import Litestar

from litestar_events.contrib.postgres import PostgresEventEmitter

app = Litestar(
    route_handlers=[register_user],
    listeners=[send_welcome_email, record_analytics],
    event_emitter_backend=partial(
        PostgresEventEmitter, dsn="postgresql://postgres:postgres@localhost:5432/events"
    ),
)

AWS SQS (durable, at-least-once, managed)

from functools import partial

from litestar import Litestar

from litestar_events.contrib.sqs import SQSEventEmitter

app = Litestar(
    route_handlers=[register_user],
    listeners=[send_welcome_email, record_analytics],
    event_emitter_backend=partial(
        SQSEventEmitter,
        queue_name="litestar_events",
        region_name="eu-central-1",
        # credentials come from the standard boto chain (env, instance role, ...)
    ),
)

GCP Pub/Sub (durable, broadcast fanout)

from functools import partial

from litestar import Litestar

from litestar_events.contrib.pubsub import PubSubEventEmitter

app = Litestar(
    route_handlers=[register_user],
    listeners=[send_welcome_email, record_analytics],
    event_emitter_backend=partial(
        PubSubEventEmitter,
        project_id="my-gcp-project",
        topic_id="litestar_events",
        # subscription_name="my-app",  # set for work-queue across replicas
    ),
)

ZeroMQ (brokerless, fire-and-forget)

from functools import partial

from litestar import Litestar

from litestar_events.contrib.zmq import ZeroMQEventEmitter

app = Litestar(
    route_handlers=[register_user],
    listeners=[send_welcome_email, record_analytics],
    event_emitter_backend=partial(
        ZeroMQEventEmitter, pub_address="tcp://127.0.0.1:5557"
    ),
)

3. Trigger an event

curl -X POST http://localhost:8000/users \
    -H 'content-type: application/json' \
    -d '{"email": "ada@example.com"}'

Both listeners run concurrently for every emit, regardless of which backend you picked. The choice of backend changes how the event gets from emitter to listener (and what happens when things go wrong), not whether the listener fires.

Examples

Runnable examples for each backend live under examples/:

• examples/rabbit — RabbitMQ via aio-pika
• examples/kafka — Kafka via aiokafka
• examples/confluent — Kafka via confluent-kafka
• examples/nats — NATS
• examples/redis — Redis Pub/Sub
• examples/mqtt — MQTT
• examples/postgres — PostgreSQL LISTEN/NOTIFY
• examples/sqs — AWS SQS (LocalStack-friendly)
• examples/pubsub — GCP Pub/Sub (emulator-friendly)
• examples/zmq — ZeroMQ (brokerless PUB/SUB)

Delivery semantics

Different backends offer different guarantees. The library tries to give each backend the strongest reasonable default for its transport, while letting you weaken or tune those guarantees via constructor arguments:

• RabbitMQ — at-least-once with per-listener error isolation and a dead-letter exchange for unparseable messages.
• Kafka / Confluent — at-least-once with consumer-group offsets.
• AWS SQS — at-least-once and durable; a message is deleted only after its listeners run. Instances sharing one queue are competing consumers (work-queue); broadcast fanout requires fronting the queue with SNS.
• GCP Pub/Sub — at-least-once and durable. Each instance gets its own subscription by default (broadcast fanout); a shared subscription_name gives work-queue semantics.
• ZeroMQ — fire-and-forget, at-most-once, brokerless. Events emitted before a subscriber finishes connecting are lost (no buffer, no persistence).
• NATS — at-most-once by default; JetStream-backed at-least-once when configured.
• Redis / MQTT / Postgres LISTEN/NOTIFY — fire-and-forget pub/sub; events emitted while no subscriber is connected are lost.

See each backend's docstring for the exact knobs.

Support ❤️

If you have any questions or need help, feel free to open an issue on the GitHub repository.

Motivation

Litestar ships with an in-process event emitter (SimpleEventEmitter) that runs listeners in the same process that emitted the event. That is great for side-effects that can tolerate "best-effort, same-process" delivery, but it falls short as soon as you need:

• Cross-process fanout — multiple app instances reacting to the same event.
• Durability — events that survive a worker crash or restart.
• Backpressure / decoupling — producers that should not wait on slow listeners.

About

litestar_events provides drop-in BaseEventEmitterBackend implementations backed by common messaging systems, so you can keep Litestar's @listener ergonomics while swapping the transport underneath.

Author 🫅

This project is maintained by Hasan Sezer Taşan, It's me 👋

Analysis

• Snyk Python Package Health Analysis
• Libraries.io - PyPI
• Safety DB
• PePy Download Stats
• PyPI Download Stats
• Pip Trends Download Stats
• PyPI Map Dependency Graph

Contributing ❤️

Any contributions are welcome! Please follow the Contributing Guidelines to contribute to this project.

Development 🧰

See the Contributing Guidelines for local setup, the common development tasks (exposed via mise), building and previewing the documentation, and the VS Code debugging configurations.

Releasing

Versioning and releases are automated with release-please, driven by Conventional Commit PR titles squash-merged into main. release-please maintains a release PR that bumps the version and CHANGELOG.md; merging it tags the release and publishes to PyPI. See the Contributing Guidelines for the commit conventions, and the Repository setup guide for one-time configuration and optional post-launch integrations such as a social preview, downstream packaging, and Repology.

Pause merges that change .github/workflows/, including Renovate action-pin updates, from merging a release PR until the entire Release run finishes. Workflow differences between the release commit and current main can block tag or release creation with a 403 when using GITHUB_TOKEN. See Release recovery for prevention and recovery steps.

For adoption or template-update reconciliation, ask your agent to "audit this template adoption/update against our existing project behavior". The shipped template-adoption skill also supports audits after an update was applied. It compares workflow behavior, required checks, custom tooling, and documentation, and asks before consequential cleanup. Its prek-workflow example shows why duplicate commands alone do not justify removing a workflow. Use the sibling repo-setup skill for repository settings and release setup.

Credits

This package was created with Copier and the hasansezertasan/copier-pyproject project template.

License 📜

This project is licensed under the MIT License.

Changelog 📝

For a detailed list of changes, please refer to the CHANGELOG.