diff --git a/asgiref/typing.py b/asgiref/typing.py new file mode 100644 index 00000000..c2ec326e --- /dev/null +++ b/asgiref/typing.py @@ -0,0 +1,202 @@ +from typing import Awaitable, Callable, Dict, Iterable, Optional, Tuple, Type, Union + +try: + from typing import Literal, Protocol, TypedDict +except ImportError: + from typing_extensions import Literal, Protocol, TypedDict # type: ignore + + +class ASGIVersions(TypedDict, total=False): + spec_version: str + version: Union[Literal["2.0"], Literal["3.0"]] + + +class HTTPScope(TypedDict): + type: Literal["http"] + asgi: ASGIVersions + http_version: str + method: str + scheme: str + path: str + raw_path: bytes + query_string: bytes + root_path: str + headers: Iterable[Tuple[bytes, bytes]] + client: Optional[Tuple[str, int]] + server: Optional[Tuple[str, Optional[int]]] + extensions: Dict[str, dict] + + +class WebsocketScope(TypedDict): + type: Literal["websocket"] + asgi: ASGIVersions + http_version: str + scheme: str + path: str + raw_path: bytes + query_string: bytes + root_path: str + headers: Iterable[Tuple[bytes, bytes]] + client: Optional[Tuple[str, int]] + server: Optional[Tuple[str, Optional[int]]] + subprotocols: Iterable[str] + extensions: Dict[str, dict] + + +class LifespanScope(TypedDict): + type: Literal["lifespan"] + asgi: ASGIVersions + + +WWWScope = Union[HTTPScope, WebsocketScope] +Scope = Union[HTTPScope, WebsocketScope, LifespanScope] + + +class HTTPRequestEvent(TypedDict): + type: Literal["http.request"] + body: bytes + more_body: bool + + +class HTTPResponseStartEvent(TypedDict): + type: Literal["http.response.start"] + status: int + headers: Iterable[Tuple[bytes, bytes]] + + +class HTTPResponseBodyEvent(TypedDict): + type: Literal["http.response.body"] + body: bytes + more_body: bool + + +class HTTPServerPushEvent(TypedDict): + type: Literal["http.response.push"] + path: str + headers: Iterable[Tuple[bytes, bytes]] + + +class HTTPDisconnectEvent(TypedDict): + type: Literal["http.disconnect"] + + +class WebsocketConnectEvent(TypedDict): + type: Literal["websocket.connect"] + + +class WebsocketAcceptEvent(TypedDict): + type: Literal["websocket.accept"] + subprotocol: Optional[str] + headers: Iterable[Tuple[bytes, bytes]] + + +class WebsocketReceiveEvent(TypedDict): + type: Literal["websocket.receive"] + bytes: Optional[bytes] + text: Optional[str] + + +class WebsocketSendEvent(TypedDict): + type: Literal["websocket.send"] + bytes: Optional[bytes] + text: Optional[str] + + +class WebsocketResponseStartEvent(TypedDict): + type: Literal["websocket.http.response.start"] + status: int + headers: Iterable[Tuple[bytes, bytes]] + + +class WebsocketResponseBodyEvent(TypedDict): + type: Literal["websocket.http.response.body"] + body: bytes + more_body: bool + + +class WebsocketDisconnectEvent(TypedDict): + type: Literal["websocket.disconnect"] + code: int + + +class WebsocketCloseEvent(TypedDict): + type: Literal["websocket.close"] + code: Optional[int] + + +class LifespanStartupEvent(TypedDict): + type: Literal["lifespan.startup"] + + +class LifespanShutdownEvent(TypedDict): + type: Literal["lifespan.shutdown"] + + +class LifespanStartupCompleteEvent(TypedDict): + type: Literal["lifespan.startup.complete"] + + +class LifespanStartupFailedEvent(TypedDict): + type: Literal["lifespan.startup.failed"] + message: str + + +class LifespanShutdownCompleteEvent(TypedDict): + type: Literal["lifespan.shutdown.complete"] + + +class LifespanShutdownFailedEvent(TypedDict): + type: Literal["lifespan.shutdown.failed"] + message: str + + +ASGIReceiveEvent = Union[ + HTTPRequestEvent, + HTTPDisconnectEvent, + WebsocketConnectEvent, + WebsocketReceiveEvent, + WebsocketDisconnectEvent, + LifespanStartupEvent, + LifespanShutdownEvent, +] + + +ASGISendEvent = Union[ + HTTPResponseStartEvent, + HTTPResponseBodyEvent, + HTTPServerPushEvent, + HTTPDisconnectEvent, + WebsocketAcceptEvent, + WebsocketSendEvent, + WebsocketResponseStartEvent, + WebsocketResponseBodyEvent, + WebsocketCloseEvent, + LifespanStartupCompleteEvent, + LifespanStartupFailedEvent, + LifespanShutdownCompleteEvent, + LifespanShutdownFailedEvent, +] + + +ASGIReceiveCallable = Callable[[], Awaitable[ASGIReceiveEvent]] +ASGISendCallable = Callable[[ASGISendEvent], Awaitable[None]] + + +class ASGI2Protocol(Protocol): + def __init__(self, scope: Scope) -> None: + ... + + async def __call__(self, receive: ASGIReceiveCallable, send: ASGISendCallable) -> None: + ... + + +ASGI2Framework = Type[ASGI2Protocol] +ASGI3Framework = Callable[ + [ + Scope, + ASGIReceiveCallable, + ASGISendCallable, + ], + Awaitable[None], +] +ASGIFramework = Union[ASGI2Framework, ASGI3Framework] diff --git a/setup.cfg b/setup.cfg index b335e298..244ba7af 100644 --- a/setup.cfg +++ b/setup.cfg @@ -31,6 +31,8 @@ project_urls = python_requires = >=3.5 packages = find: include_package_data = true +install_requires = + typing_extensions; python_version < "3.8" zip_safe = false [options.extras_require]