Registry / web-framework / fastapi-events

fastapi-events

JSON →
library0.12.2pypypi✓ verified 85d ago

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
INSTALL
IMPORT
SIG · FASTAPI-EVENTS
F
fastapi-events
web-frameworkpythonv0.12.2
Install
1.9s avg
Import
Disk
50MB
Pass rate
10/ 10
Env Coverage10 / 10
glibc
3.93.13
musl
3.93.13
Install & Compatibility
Where this runs
tested against v0.12.2 · pip install
no network on importno background threads
Install × environment matrix
Each cell = how many times install + import succeeded across repeated harness runs. Partial = flaky.
glibc = Debian/Ubuntu slim · musl = Alpine Linux
musl
py 3.103.925 runs
installs and imports cleanly · install 0.0s · import 0.000s · 51.2MB
glibc
py 3.103.925 runs
installs and imports cleanly · install 1.9s · import 0.000s · 52MB
50MB installed
● package 50MB
Code
Verified usage

Verified import paths — ran on the pinned version, not inferred.

BaseEventHandler
from fastapi_events import BaseEventHandler
from fastapi_events import event_dispatch
event_store
from fastapi_events import event_store
from fastapi_events import event_dispatch
handler_store
from fastapi_events import handler_store
from fastapi_events import event_dispatch

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}")
Debug
Known issues
deprecatedImplicit 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.
affects: >=0.10.0
breakingPydantic 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.
affects: >=0.11.0
gotchaEvent 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).
affects: all
gotchaMiddleware 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.
affects: all
Upgrade
Version history
0.12.2latest on PyPI · released Dec 21, 2024
Audit
Dependencies
fastapirequiredRequired; the library is built on top of FastAPI
pydanticrequiredRequired for event schema definition and validation
boto3optionalRequired for AWS SQS backend
google-cloud-pubsuboptionalRequired for Google Cloud Pub/Sub backend
aio-pikaoptionalRequired for RabbitMQ backend
redisoptionalRequired for Redis backend
opentelemetry-apioptionalRequired for OpenTelemetry integration
Agent activity
41 hits · last 30 days
node
36
OpenAI (training)
2
Resources
fastapi-events — pip install fastapi-events · libregistry