Last active
December 14, 2023 09:56
-
-
Save WilliamStam/ecc72ef53091ff42c133fc9c169ef88d to your computer and use it in GitHub Desktop.
Fastapi help needed - event system
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
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