Move builtin signals to enum (#2309)

* Move builtin signals to enum

* Fix annotations
This commit is contained in:
Adam Hopkins 2021-11-14 23:21:14 +02:00 committed by GitHub
parent 392a497366
commit 9a9f72ad64
No known key found for this signature in database
GPG Key ID: 4AEE18F83AFDEB23
4 changed files with 65 additions and 26 deletions

View File

@ -4,6 +4,7 @@ import asyncio
from collections import defaultdict
from copy import deepcopy
from enum import Enum
from types import SimpleNamespace
from typing import (
TYPE_CHECKING,
@ -144,7 +145,7 @@ class Blueprint(BaseSanic):
kwargs["apply"] = False
return super().exception(*args, **kwargs)
def signal(self, event: str, *args, **kwargs):
def signal(self, event: Union[str, Enum], *args, **kwargs):
kwargs["apply"] = False
return super().signal(event, *args, **kwargs)

View File

@ -1,4 +1,5 @@
from typing import Any, Callable, Dict, Optional, Set
from enum import Enum
from typing import Any, Callable, Dict, Optional, Set, Union
from sanic.models.futures import FutureSignal
from sanic.models.handler_types import SignalHandler
@ -19,7 +20,7 @@ class SignalMixin:
def signal(
self,
event: str,
event: Union[str, Enum],
*,
apply: bool = True,
condition: Dict[str, Any] = None,
@ -41,13 +42,11 @@ class SignalMixin:
filtering, defaults to None
:type condition: Dict[str, Any], optional
"""
event_value = str(event.value) if isinstance(event, Enum) else event
def decorator(handler: SignalHandler):
nonlocal event
nonlocal apply
future_signal = FutureSignal(
handler, event, HashableDict(condition or {})
handler, event_value, HashableDict(condition or {})
)
self._future_signals.add(future_signal)

View File

@ -2,6 +2,7 @@ from __future__ import annotations
import asyncio
from enum import Enum
from inspect import isawaitable
from typing import Any, Dict, List, Optional, Tuple, Union
@ -14,29 +15,47 @@ from sanic.log import error_logger, logger
from sanic.models.handler_types import SignalHandler
class Event(Enum):
SERVER_INIT_AFTER = "server.init.after"
SERVER_INIT_BEFORE = "server.init.before"
SERVER_SHUTDOWN_AFTER = "server.shutdown.after"
SERVER_SHUTDOWN_BEFORE = "server.shutdown.before"
HTTP_LIFECYCLE_BEGIN = "http.lifecycle.begin"
HTTP_LIFECYCLE_COMPLETE = "http.lifecycle.complete"
HTTP_LIFECYCLE_EXCEPTION = "http.lifecycle.exception"
HTTP_LIFECYCLE_HANDLE = "http.lifecycle.handle"
HTTP_LIFECYCLE_READ_BODY = "http.lifecycle.read_body"
HTTP_LIFECYCLE_READ_HEAD = "http.lifecycle.read_head"
HTTP_LIFECYCLE_REQUEST = "http.lifecycle.request"
HTTP_LIFECYCLE_RESPONSE = "http.lifecycle.response"
HTTP_ROUTING_AFTER = "http.routing.after"
HTTP_ROUTING_BEFORE = "http.routing.before"
HTTP_LIFECYCLE_SEND = "http.lifecycle.send"
HTTP_MIDDLEWARE_AFTER = "http.middleware.after"
HTTP_MIDDLEWARE_BEFORE = "http.middleware.before"
RESERVED_NAMESPACES = {
"server": (
# "server.main.start",
# "server.main.stop",
"server.init.before",
"server.init.after",
"server.shutdown.before",
"server.shutdown.after",
Event.SERVER_INIT_AFTER.value,
Event.SERVER_INIT_BEFORE.value,
Event.SERVER_SHUTDOWN_AFTER.value,
Event.SERVER_SHUTDOWN_BEFORE.value,
),
"http": (
"http.lifecycle.begin",
"http.lifecycle.complete",
"http.lifecycle.exception",
"http.lifecycle.handle",
"http.lifecycle.read_body",
"http.lifecycle.read_head",
"http.lifecycle.request",
"http.lifecycle.response",
"http.routing.after",
"http.routing.before",
"http.lifecycle.send",
"http.middleware.after",
"http.middleware.before",
Event.HTTP_LIFECYCLE_BEGIN.value,
Event.HTTP_LIFECYCLE_COMPLETE.value,
Event.HTTP_LIFECYCLE_EXCEPTION.value,
Event.HTTP_LIFECYCLE_HANDLE.value,
Event.HTTP_LIFECYCLE_READ_BODY.value,
Event.HTTP_LIFECYCLE_READ_HEAD.value,
Event.HTTP_LIFECYCLE_REQUEST.value,
Event.HTTP_LIFECYCLE_RESPONSE.value,
Event.HTTP_ROUTING_AFTER.value,
Event.HTTP_ROUTING_BEFORE.value,
Event.HTTP_LIFECYCLE_SEND.value,
Event.HTTP_MIDDLEWARE_AFTER.value,
Event.HTTP_MIDDLEWARE_BEFORE.value,
),
}

View File

@ -1,5 +1,6 @@
import asyncio
from enum import Enum
from inspect import isawaitable
import pytest
@ -50,6 +51,25 @@ def test_invalid_signal(app, signal):
...
@pytest.mark.asyncio
async def test_dispatch_signal_with_enum_event(app):
counter = 0
class FooEnum(Enum):
FOO_BAR_BAZ = "foo.bar.baz"
@app.signal(FooEnum.FOO_BAR_BAZ)
def sync_signal(*_):
nonlocal counter
counter += 1
app.signal_router.finalize()
await app.dispatch("foo.bar.baz")
assert counter == 1
@pytest.mark.asyncio
async def test_dispatch_signal_triggers_multiple_handlers(app):
counter = 0