Skip to content

Instantly share code, notes, and snippets.

@WilliamStam
Last active December 14, 2023 09:56
Show Gist options
  • Save WilliamStam/ecc72ef53091ff42c133fc9c169ef88d to your computer and use it in GitHub Desktop.
Save WilliamStam/ecc72ef53091ff42c133fc9c169ef88d to your computer and use it in GitHub Desktop.
Fastapi help needed - event system
import dataclasses
import logging
from typing import Annotated, Callable, Dict, List
from fastapi import Depends, FastAPI
logger = logging.getLogger(__name__)
class EventsCollection():
def __init__(self):
self.events: Dict[str, List[Callable]] = {}
def register(self, event, listener):
event_name = f"{event.__module__}.{event.__name__}"
logger.warning(f"registering event [{event_name}] with listener [{str(listener)}]")
self.events.setdefault(event_name, [])
self.events[event_name].append(listener)
async def dispatch(self, event):
event_name = f"{event.__module__}.{event.__class__.__name__}"
logger.warning(f"dispatching event [{event_name}]")
listeners = self.events.get(event_name, None)
if listeners:
for listener in listeners:
# listener needs to be able to request whatever defined dependency there is here like DB / user / request etc as well as pasing the event object through
await listener(event)
events = EventsCollection()
#
#
# Events = Annotated[EventsCollection, Depends(events)]
# want to pass this event via dispatch
@dataclasses.dataclass
class BasicEvent:
bird: str
owl: str
# want this to be looked up in the listener for the event
@dataclasses.dataclass
class ListenerDependency:
dog: str
cat: str
# defining the listeners dependency
async def basic_dependency():
return ListenerDependency(
dog="woof",
cat="meow"
)
BasicDependency = Annotated[ListenerDependency, Depends(basic_dependency)]
# the listener
async def listener(event, dep: BasicDependency):
print("LISTENER CALLED")
print(event)
print(dep)
# registering the event
events.register(BasicEvent, listener)
app = FastAPI()
# works but no event
@app.get("/")
async def testing(dep: BasicDependency):
print(dep)
# ListenerDependency(dog='woof', cat='meow')
# want this to work
@app.get("/")
async def testing(events: Events):
event = BasicEvent(
bird="squeak",
owl="hiss"
)
events.dispatch(event)
# desired
# BasicEvent(bird='squeak',owl='hiss')
# ListenerDependency(dog='woof', cat='meow')
#
# since the event got dispatched, so the listener outputs the event, and then its own defined dependency
# the idea here is that i only need to pass events to the service layer instead of every single possible dependency there might be, user / db / request etc
Sign up for free to join this conversation on GitHub. Already have an account? Sign in to comment