fastapi-events

0.12.2 verified Mon Apr 27 auth: no python

Event dispatching library for FastAPI that supports synchronous and asynchronous event handlers, multiple backends (AWS SQS, Google Cloud Pub/Sub, RabbitMQ, Redis, etc.), OpenTelemetry integration, and Pydantic model dispatching. Current version 0.12.2, release cadence ~monthly.

pip install fastapi-events

Common errors

error ImportError: cannot import name 'event_dispatch' from 'fastapi_events.dispatcher' ↓

cause Wrong import path; event_dispatch is directly under fastapi_events.

fix

Use: from fastapi_events import event_dispatch

error AttributeError: module 'fastapi_events.handlers' has no attribute 'LocalHandler' ↓

cause LocalHandler moved to fastapi_events.handlers.local in version 0.9.0.

fix

Use: from fastapi_events.handlers.local import LocalHandler

error DeprecationWarning: The 'dict' method is deprecated; use 'model_dump' instead ↓

cause Pydantic v2 deprecation; fastapi-events >=0.10.2 emits this warning.

fix

Upgrade to fastapi-events >=0.11.0 or configure Pydantic v2 to use model_dump.

error fastapi.exceptions.FastAPIError: Invalid args for response. Response must be a Starlette Response, or a subclass thereof. ↓

cause Event handler returns an HTTP response object instead of None; handlers should not return responses.

fix

Make sure event handlers return None (implicitly) and do not return FastAPI Response objects.

Warnings

deprecated Implicit event name from function name as string literal is deprecated in 0.10.0+; use explicit event_name argument. ↓

fix Use event_dispatch('event_name', payload=...) instead of relying on function name.

breaking Pydantic v1 .dict() method is deprecated and removed in Pydantic v2; events using Pydantic models require model_dump(). ↓

fix Use .model_dump() for Pydantic v2 models, or set config to opt-out of schema dump.

gotcha Event handlers registered as coroutines must be awaited; mixing sync and async handlers can cause missing events. ↓

fix Ensure handler function signature matches the dispatcher context (sync vs async).

gotcha Middleware order matters: EventHandlerASGIMiddleware must be added before other middleware that may intercept events. ↓

fix Add EventHandlerASGIMiddleware as the first middleware using app.add_middleware() at the top.

Install

pip install fastapi-events[aws]

pip install fastapi-events[gcp]

pip install fastapi-events[rabbit]

pip install fastapi-events[redis]

Imports

event_dispatch

wrong

from fastapi_events.dispatcher import event_dispatch

correct

from fastapi_events import event_dispatch

Wrong import path; the dispatcher is directly under fastapi_events

EventHandler

wrong

from fastapi_events.handlers import LocalHandler

correct

from fastapi_events.handlers.local import LocalHandler

Pre-0.9.0 pattern; LocalHandler moved to .local submodule

Payload

wrong

from fastapi_events.models import Payload

correct

from fastapi_events.typing import EventModel

Model schema classes were renamed in 0.10.0

Quickstart

Basic FastAPI app that dispatches a local event and handles it synchronously.

from fastapi import FastAPI
from fastapi_events import event_dispatch
from fastapi_events.middleware import EventHandlerASGIMiddleware
from fastapi_events.handlers.local import LocalHandler

app = FastAPI()
handler = LocalHandler()

app.add_middleware(
    EventHandlerASGIMiddleware,
    handlers=[handler],
)

@app.get("/")
async def root():
    # Dispatch an event that will be handled locally
    event_dispatch("my_event", payload={"key": "value"})
    return {"message": "Event dispatched"}

@handler.register("my_event")
def handle_my_event(payload: dict):
    print(f"Received event with payload: {payload}")