Here is a super simple implementation of the new router as the beginning of the signals API:
from sanic_routing import BaseRouter, Route
class Signal(Route):
def get_handler(self, raw_path, method, _):
method = method or self.router.DEFAULT_METHOD
raw_path = raw_path.lstrip(self.router.delimiter)
try:
return self.handlers[raw_path][method]
except (IndexError, KeyError):
raise self.router.method_handler_exception(
f"Method '{method}' not found on {self}",
method=method,
allowed_methods=set(self.methods[raw_path]),
)
class SignalRouter(BaseRouter):
def __init__(self) -> None:
super().__init__(
delimiter=".",
route_class=Signal,
stacking=True,
)
def get(self, event: str):
return self.resolve(f".{event}")
def send(self, event: str) -> None:
signal, handlers, params = self.get(event)
for handler in handlers:
handler(**params)
def callback1():
print("called1")
def callback2():
print("called2")
signals = SignalRouter()
signals.add("server.init.before", callback1)
signals.add("server.init.before", callback2)
signals.finalize()
signals.send("server.init.before")
See RFC : Signals Support for Sanic Proposal · Issue #1630 · sanic-org/sanic · GitHub