Awesome
fastapi-event
fastapi-event is event dispatcher for FastAPI framework.
Installation
pip3 install fastapi-event
Usage
Make Event
from fastapi_event import BaseEvent
class TestEvent(BaseEvent):
async def run(self, parameter=None):
...
Inherit BaseEvent
and override run()
method.
from fastapi_event import BaseEvent
class FirstEvent(BaseEvent):
ORDER = 1 # HERE(Optional)
async def run(self, parameter=None):
...
class SecondEvent(BaseEvent):
ORDER = 2 # HERE(Optional)
async def run(self, parameter=None):
...
If you want to determine the order between events, specify ORDER
in your event.
Then, regardless of the order in which the events are stored, they will be executed in the order specified in ORDER
variable.
However, ORDER
does not work when run_at_once=True
.
Parameter(optional)
from pydantic import BaseModel
class TestEventParameter(BaseModel):
id: str
pw: str
In case of need parameter, you have to inherit BaseModel
and set fields.
Middleware
from fastapi import FastAPI
from fastapi_event import EventHandlerMiddleware
app = FastAPI()
app.add_middleware(EventHandlerMiddleware)
EventListener
from fastapi_event import EventListener
@EventListener()
async def test():
...
Set @EventListener()
decorator on the function that emits the event.
@EventListener(run_at_once=False)
If you pass run_at_once=False
, it will execute in the order in which store()
is called. (or according to the ORDER
variable defined in the event)
Otherwise, it will execute through asyncio.gather()
to run at once.
Store event
from fastapi_event import EventListener, event_handler
@EventListener()
async def test():
await event_handler.store(
event=TestEvent,
parameter=TestParameter(id="hide", pw="hide"), # Optional
)
Store your event to handler via store()
method. (parameter is optional)
An event will be emitted after the function has finished executing.