b1b12e004e
* Update some tests * Resolve #2122 route decorator returning tuple * Use rc sanic-routing version * Update unit tests to <:str> * Minimal working version with some signals implemented * Add more http signals * Update ASGI and change listeners to signals * Allow for dynamic ODE signals * Allow signals to be stacked * Begin tests * Prioritize match_info on keyword argument injection * WIP on tests * Compat with signals * Work through some test coverage * Passing tests * Post linting * Setup proper resets * coverage reporting * Fixes from vltr comments * clear delayed tasks * Fix bad test * rm pycache
78 lines
2.1 KiB
Python
78 lines
2.1 KiB
Python
from typing import Any, Callable, Dict, Optional, Set
|
|
|
|
from sanic.models.futures import FutureSignal
|
|
from sanic.models.handler_types import SignalHandler
|
|
from sanic.signals import Signal
|
|
|
|
|
|
class HashableDict(dict):
|
|
def __hash__(self):
|
|
return hash(tuple(sorted(self.items())))
|
|
|
|
|
|
class SignalMixin:
|
|
def __init__(self, *args, **kwargs) -> None:
|
|
self._future_signals: Set[FutureSignal] = set()
|
|
|
|
def _apply_signal(self, signal: FutureSignal) -> Signal:
|
|
raise NotImplementedError # noqa
|
|
|
|
def signal(
|
|
self,
|
|
event: str,
|
|
*,
|
|
apply: bool = True,
|
|
condition: Dict[str, Any] = None,
|
|
) -> Callable[[SignalHandler], SignalHandler]:
|
|
"""
|
|
For creating a signal handler, used similar to a route handler:
|
|
|
|
.. code-block:: python
|
|
|
|
@app.signal("foo.bar.<thing>")
|
|
async def signal_handler(thing, **kwargs):
|
|
print(f"[signal_handler] {thing=}", kwargs)
|
|
|
|
:param event: Representation of the event in ``one.two.three`` form
|
|
:type event: str
|
|
:param apply: For lazy evaluation, defaults to True
|
|
:type apply: bool, optional
|
|
:param condition: For use with the ``condition`` argument in dispatch
|
|
filtering, defaults to None
|
|
:type condition: Dict[str, Any], optional
|
|
"""
|
|
|
|
def decorator(handler: SignalHandler):
|
|
nonlocal event
|
|
nonlocal apply
|
|
|
|
future_signal = FutureSignal(
|
|
handler, event, HashableDict(condition or {})
|
|
)
|
|
self._future_signals.add(future_signal)
|
|
|
|
if apply:
|
|
self._apply_signal(future_signal)
|
|
|
|
return handler
|
|
|
|
return decorator
|
|
|
|
def add_signal(
|
|
self,
|
|
handler: Optional[Callable[..., Any]],
|
|
event: str,
|
|
condition: Dict[str, Any] = None,
|
|
):
|
|
if not handler:
|
|
|
|
async def noop():
|
|
...
|
|
|
|
handler = noop
|
|
self.signal(event=event, condition=condition)(handler)
|
|
return handler
|
|
|
|
def event(self, event: str):
|
|
raise NotImplementedError
|