Publish and subscribe to events

Install the events extra:

pip install 'fastsprout[events]'

EventBus delivers events in memory. In this example, pub publishes the event returned by an action, and sub registers a handler for that event type. Both decorators use the same bus instance.

"""Publish an event and wait for its subscribed handler."""

import asyncio

from fastsprout.events import Event, EventBus, pub, sub

bus = EventBus()


class UserCreated(Event):
    email: str


@sub(UserCreated, bus=bus)
async def send_welcome(event: UserCreated) -> None:
    print(f"Welcome sent to {event.email}")


@pub(bus=bus)
async def register_user() -> UserCreated:
    return UserCreated(email="[email protected]")


async def main() -> None:
    event = await register_user()
    state = await bus.get(event)
    if state is None:
        raise RuntimeError("The event was not published")
    await state.wait()


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

Run the source file from the repository root:

uv run --extra events python docs_src/events/pub_sub.py

Result

Welcome sent to [email protected]

See the events examples for publishers, subscriber groups, scheduling, and middleware.