Event dispatching library for FastAPI
Project description
fastapi-events
An event dispatching/handling library for FastAPI, and Starlette.
Features:
- straightforward and simple API to emit events anywhere in your code
- handling of events will be done after responses are returned (doesn't affect response time)
- powerful built-in event handlers to enable customizations
- async functions are first-class citizen
Installation
pip install fastapi-events
To use it with AWS handlers, install:
pip install fastapi-events[aws]
Usage
fastapi-events
supports both FastAPI and Starlette. To use it, simply configure it as middleware.
-
Configuring
fastapi-events
for FastAPI:from fastapi import FastAPI from fastapi.requests import Request from fastapi.responses import JSONResponse from fastapi_events.dispatcher import dispatch from fastapi_events.middleware import EventHandlerASGIMiddleware from fastapi_events.handlers.local import local_handler app = FastAPI() app.add_middleware(EventHandlerASGIMiddleware, handlers=[local_handler]) # registering handler(s) @app.get("/") def index(request: Request) -> JSONResponse: dispatch("my-fancy-event", payload={"id": 1}) # Emit events anywhere in your code return JSONResponse()
-
Configuring
fastapi-events
for Starlette:from starlette.applications import Starlette from starlette.middleware import Middleware from starlette.requests import Request from starlette.responses import JSONResponse from fastapi_events.dispatcher import dispatch from fastapi_events.handlers.local import local_handler from fastapi_events.middleware import EventHandlerASGIMiddleware app = Starlette(middleware=[ Middleware(EventHandlerASGIMiddleware, handlers=[local_handler]) # registering handlers ]) @app.route("/") async def root(request: Request) -> JSONResponse: dispatch("new event", payload={"id": 1}) # Emit events anywhere in your code return JSONResponse()
Dispatching events
Events can be dispatched anywhere in the code, as long as they are dispatched before a response is made.
# anywhere in code
from fastapi_events.dispatcher import dispatch
dispatch(
"cat-requested-a-fish", # Event name, accepts any valid string
payload={"cat_id": "fd375d23-b0c9-4271-a9e0-e028c4cd7230"} # Event payload, accepts any arbitrary data
)
dispatch("a_cat_is_spotted") # This works too!
Handling Events
Handle events locally
The flexibility of fastapi-events
allows us to customise how the events should be handled.
For starter, you might want to handle your events locally.
# ex: in handlers.py
from fastapi_events.handlers.local import local_handler
from fastapi_events.typing import Event
@local_handler.register(event_name="cat*")
def handle_all_cat_events(event: Event):
"""
this handler will match with an events prefixed with `cat`.
ex: "cat_eats_a_fish", "cat_is_cute", etc
"""
# the `event` argument is nothing more than a tuple of event name and payload
event_name, payload = event
# TODO do anything you'd like with the event
@local_handler.register(event_name="cat*") # Tip: You can register several handlers with the same event name
def handle_all_cat_events_another_way(event: Event):
pass
@local_handler.register(event_name="*")
async def handle_all_events(event: Event):
# event handlers can be coroutine function too (`async def`)
pass
Handling events remotely
For larger projects, you might have services dedicated to handle events separately.
For instance, fastapi-events
comes with AWS SQS forwarder to can forward the events to a remote queue.
-
Register
SQSForwardHandler
as handlers:app = FastAPI() app.add_middleware(EventHandlerASGIMiddleware, handlers=[SQSForwardHandler(queue_url="test-queue", region_name="eu-central-1")]) # registering handler(s)
-
Start dispatching events! Events will be serialised in JSON by default:
["event name", {"payload": "here is the payload"}]
Built-in handlers
Here is a list of built-in event handlers:
-
LocalHandler
/local_handler
:- import from
fastapi_events.handlers.local
- for handle events locally. See examples above
- event name pattern matching is done using Unix shell-style matching (
fnmatch
)
- import from
-
SQSForwardHandler
:- import from
fastapi_events.handlers.aws
- forwards events to an AWS SQS queue
- import from
-
EchoHandler
:- forward events to stdout with
pprint
. Great for debugging purpose
- forward events to stdout with
Creating your own handler
Creating your own handler is nothing more than inheriting from the BaseEventHandler
in fastapi_events.handlers.base
.
To handle events, fastapi_events
calls one of these methods, in the following priority order:
-
handle_many(events)
: The coroutine function should expect the backlog of events collected. -
handle(event)
: In cases wherehandle_many()
wasn't defined in your custom handler,handle()
will be called by iterating through the events in the backlog.
from typing import Iterable
from fastapi_events.typing import Event
from fastapi_events.handlers.base import BaseEventHandler
class MyOwnEventHandler(BaseEventHandler):
async def handle(self, event: Event) -> None:
"""
Handle events one by one
"""
pass
async def handle_many(self, events: Iterable[Event]) -> None:
"""
Handle events by batch
"""
pass
Feedback, Questions?
Any form of feedback and questions are welcome! Please create an issue here.
Project details
Release history Release notifications | RSS feed
Download files
Download the file for your platform. If you're not sure which to choose, learn more about installing packages.
Source Distribution
Built Distribution
Hashes for fastapi_events-0.1-py3-none-any.whl
Algorithm | Hash digest | |
---|---|---|
SHA256 | f75663dd507c5fb47114dbd5eb3b28ec58fb2512148ef8dc0894833613a62ad6 |
|
MD5 | 26189cc332fec9cc688dce12665cf5a9 |
|
BLAKE2b-256 | cb7dd1226854d2d4ccd6345ff8c74c4bd122890afb61548a48b4026925cd1ebb |