fastapi-events fastapi 异步事件分发处理扩展

fastapi-events fastapi 异步事件分发处理扩展，提供了本地，以及远程消息处理能力，同时包含了一些内置的handler，对于自定义handler 也是比较灵活的

参考使用

• app.py

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()    
自定义handler 处理
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

from fastapi.requests import Request

说明

在项目中使用此模块可以实现异步的消息处理，实现灵活的业务处理

参考资料

https://github.com/melvinkcx/fastapi-events