Add typing for the ASGI messages

This makes use of the TypedDict (added to Python in 3.8 and available
in the typing_extensions package) to specify the types of the key
values in the ASGI messages. This then ensures that the messages are
correctly constructued and used in the code.

Note the type ignores are due to this issue
python/mypy#8533.

src/hypercorn/asyncio/context.py

@@ -1,9 +1,16 @@
 import asyncio
-from typing import Any, Awaitable, Callable, Type, Union
+from typing import Any, Awaitable, Callable, Optional, Type, Union
 
 from .task_group import TaskGroup
 from ..config import Config
-from ..typing import ASGIFramework, Event, Scope
+from ..typing import (
+    ASGIFramework,
+    ASGIReceiveCallable,
+    ASGIReceiveEvent,
+    ASGISendEvent,
+    Event,
+    Scope,
+)
 from ..utils import invoke_asgi
 
 
@@ -22,7 +29,11 @@ async def set(self) -> None:
 
 
 async def _handle(
-    app: ASGIFramework, config: Config, scope: Scope, receive: Callable, send: Callable
+    app: ASGIFramework,
+    config: Config,
+    scope: Scope,
+    receive: ASGIReceiveCallable,
+    send: Callable[[Optional[ASGISendEvent]], Awaitable[None]],
 ) -> None:
     try:
         await invoke_asgi(app, scope, receive, send)
@@ -45,9 +56,9 @@ async def spawn_app(
         app: ASGIFramework,
         config: Config,
         scope: Scope,
-        send: Callable[[dict], Awaitable[None]],
-    ) -> Callable[[dict], Awaitable[None]]:
-        app_queue: asyncio.Queue = asyncio.Queue(config.max_app_queue_size)
+        send: Callable[[Optional[ASGISendEvent]], Awaitable[None]],
+    ) -> Callable[[ASGIReceiveEvent], Awaitable[None]]:
+        app_queue: asyncio.Queue[ASGIReceiveEvent] = asyncio.Queue(config.max_app_queue_size)
         self.task_group.spawn(_handle(app, config, scope, app_queue.get, send))
         return app_queue.put
 

src/hypercorn/asyncio/lifespan.py

@@ -1,7 +1,7 @@
 import asyncio
 
 from ..config import Config
-from ..typing import ASGIFramework, LifespanScope
+from ..typing import ASGIFramework, ASGIReceiveEvent, ASGISendEvent, LifespanScope
 from ..utils import invoke_asgi, LifespanFailure, LifespanTimeout
 
 
@@ -67,10 +67,10 @@ async def wait_for_shutdown(self) -> None:
         except asyncio.TimeoutError as error:
             raise LifespanTimeout("shutdown") from error
 
-    async def asgi_receive(self) -> dict:
+    async def asgi_receive(self) -> ASGIReceiveEvent:
         return await self.app_queue.get()
 
-    async def asgi_send(self, message: dict) -> None:
+    async def asgi_send(self, message: ASGISendEvent) -> None:
         if message["type"] == "lifespan.startup.complete":
             self.startup.set()
         elif message["type"] == "lifespan.shutdown.complete":

src/hypercorn/logging.py

@@ -8,7 +8,7 @@
 
 if TYPE_CHECKING:
     from .config import Config
-    from .typing import WWWScope
+    from .typing import ResponseSummary, WWWScope
 
 
 def _create_logger(
@@ -65,7 +65,9 @@ def __init__(self, config: "Config") -> None:
             if config.logconfig_dict is not None:
                 dictConfig(config.logconfig_dict)
 
-    async def access(self, request: "WWWScope", response: dict, request_time: float) -> None:
+    async def access(
+        self, request: "WWWScope", response: "ResponseSummary", request_time: float
+    ) -> None:
         if self.access_logger is not None:
             self.access_logger.info(
                 self.access_log_format, self.atoms(request, response, request_time)
@@ -99,7 +101,9 @@ async def log(self, level: int, message: str, *args: Any, **kwargs: Any) -> None
         if self.error_logger is not None:
             self.error_logger.log(level, message, *args, **kwargs)
 
-    def atoms(self, request: "WWWScope", response: dict, request_time: float) -> Mapping[str, str]:
+    def atoms(
+        self, request: "WWWScope", response: "ResponseSummary", request_time: float
+    ) -> Mapping[str, str]:
         """Create and return an access log atoms dictionary.
 
         This can be overidden and customised if desired. It should
@@ -112,7 +116,9 @@ def __getattr__(self, name: str) -> Any:
 
 
 class AccessLogAtoms(dict):
-    def __init__(self, request: "WWWScope", response: dict, request_time: float) -> None:
+    def __init__(
+        self, request: "WWWScope", response: "ResponseSummary", request_time: float
+    ) -> None:
         for name, value in request["headers"]:
             self[f"{{{name.decode('latin1').lower()}}}i"] = value.decode("latin1")
         for name, value in response.get("headers", []):

src/hypercorn/protocol/http_stream.py

@@ -5,7 +5,7 @@
 
 from .events import Body, EndBody, Event, Request, Response, StreamClosed
 from ..config import Config
-from ..typing import ASGIFramework, Context, HTTPScope
+from ..typing import ASGIFramework, ASGISendEvent, Context, HTTPResponseStartEvent, HTTPScope
 from ..utils import build_and_validate_headers, suppress_body, UnexpectedMessage, valid_server_name
 
 PUSH_VERSIONS = {"2", "3"}
@@ -37,7 +37,7 @@ def __init__(
         self.closed = False
         self.config = config
         self.context = context
-        self.response: dict
+        self.response: HTTPResponseStartEvent
         self.scope: HTTPScope
         self.send = send
         self.scheme = "https" if ssl else "http"
@@ -91,9 +91,9 @@ async def handle(self, event: Event) -> None:
         elif isinstance(event, StreamClosed):
             self.closed = True
             if self.app_put is not None:
-                await self.app_put({"type": "http.disconnect"})
+                await self.app_put({"type": "http.disconnect"})  # type: ignore
 
-    async def app_send(self, message: Optional[dict]) -> None:
+    async def app_send(self, message: Optional[ASGISendEvent]) -> None:
         if self.closed:
             # Allow app to finish after close
             return

src/hypercorn/protocol/ws_stream.py

@@ -19,7 +19,15 @@
 
 from .events import Body, Data, EndBody, EndData, Event, Request, Response, StreamClosed
 from ..config import Config
-from ..typing import ASGIFramework, Context, WebsocketScope
+from ..typing import (
+    ASGIFramework,
+    ASGISendEvent,
+    Context,
+    WebsocketAcceptEvent,
+    WebsocketResponseBodyEvent,
+    WebsocketResponseStartEvent,
+    WebsocketScope,
+)
 from ..utils import build_and_validate_headers, suppress_body, UnexpectedMessage, valid_server_name
 
 
@@ -152,7 +160,7 @@ def __init__(
         self.closed = False
         self.config = config
         self.context = context
-        self.response: dict
+        self.response: WebsocketResponseStartEvent
         self.scope: WebsocketScope
         self.send = send
         # RFC 8441 for HTTP/2 says use http or https, ASGI says ws or wss
@@ -202,7 +210,7 @@ async def handle(self, event: Event) -> None:
                 self.app_put = await self.context.spawn_app(
                     self.app, self.config, self.scope, self.app_send
                 )
-                await self.app_put({"type": "websocket.connect"})
+                await self.app_put({"type": "websocket.connect"})  # type: ignore
         elif isinstance(event, (Body, Data)):
             self.connection.receive_data(event.data)
             await self._handle_events()
@@ -215,7 +223,7 @@ async def handle(self, event: Event) -> None:
                     code = CloseReason.ABNORMAL_CLOSURE.value
                 await self.app_put({"type": "websocket.disconnect", "code": code})
 
-    async def app_send(self, message: Optional[dict]) -> None:
+    async def app_send(self, message: Optional[ASGISendEvent]) -> None:
         if self.closed:
             # Allow app to finish after close
             return
@@ -304,7 +312,7 @@ async def _send_wsproto_event(self, event: WSProtoEvent) -> None:
         data = self.connection.send(event)
         await self.send(Data(stream_id=self.stream_id, data=data))
 
-    async def _accept(self, message: dict) -> None:
+    async def _accept(self, message: WebsocketAcceptEvent) -> None:
         self.state = ASGIWebsocketState.CONNECTED
         status_code, headers, self.connection = self.handshake.accept(message.get("subprotocol"))
         await self.send(
@@ -316,7 +324,7 @@ async def _accept(self, message: dict) -> None:
         if self.config.websocket_ping_interval is not None:
             self.context.spawn(self._send_pings)
 
-    async def _send_rejection(self, message: dict) -> None:
+    async def _send_rejection(self, message: WebsocketResponseBodyEvent) -> None:
         body_suppressed = suppress_body("GET", self.response["status"])
         if self.state == ASGIWebsocketState.HANDSHAKE:
             headers = build_and_validate_headers(self.response["headers"])

src/hypercorn/statsd.py

@@ -4,7 +4,7 @@
 
 if TYPE_CHECKING:
     from .config import Config
-    from .typing import WWWScope
+    from .typing import ResponseSummary, WWWScope
 
 METRIC_VAR = "metric"
 VALUE_VAR = "value"
@@ -64,7 +64,9 @@ async def log(self, level: int, message: str, *args: Any, **kwargs: Any) -> None
         except Exception:
             await super().warning("Failed to log to statsd", exc_info=True)
 
-    async def access(self, request: "WWWScope", response: dict, request_time: float) -> None:
+    async def access(
+        self, request: "WWWScope", response: "ResponseSummary", request_time: float
+    ) -> None:
         await super().access(request, response, request_time)
         await self.histogram("hypercorn.request.duration", request_time * 1_000)
         await self.increment("hypercorn.requests", 1)

src/hypercorn/trio/context.py

@@ -1,9 +1,16 @@
-from typing import Any, Awaitable, Callable, Type, Union
+from typing import Any, Awaitable, Callable, Optional, Type, Union
 
 import trio
 
 from ..config import Config
-from ..typing import ASGIFramework, Event, Scope
+from ..typing import (
+    ASGIFramework,
+    ASGIReceiveCallable,
+    ASGIReceiveEvent,
+    ASGISendEvent,
+    Event,
+    Scope,
+)
 from ..utils import invoke_asgi
 
 
@@ -22,7 +29,11 @@ async def set(self) -> None:
 
 
 async def _handle(
-    app: ASGIFramework, config: Config, scope: Scope, receive: Callable, send: Callable
+    app: ASGIFramework,
+    config: Config,
+    scope: Scope,
+    receive: ASGIReceiveCallable,
+    send: Callable[[Optional[ASGISendEvent]], Awaitable[None]],
 ) -> None:
     try:
         await invoke_asgi(app, scope, receive, send)
@@ -54,8 +65,8 @@ async def spawn_app(
         app: ASGIFramework,
         config: Config,
         scope: Scope,
-        send: Callable[[dict], Awaitable[None]],
-    ) -> Callable[[dict], Awaitable[None]]:
+        send: Callable[[Optional[ASGISendEvent]], Awaitable[None]],
+    ) -> Callable[[ASGIReceiveEvent], Awaitable[None]]:
         app_send_channel, app_receive_channel = trio.open_memory_channel(config.max_app_queue_size)
         self.nursery.start_soon(_handle, app, config, scope, app_receive_channel.receive, send)
         return app_send_channel.send

src/hypercorn/trio/lifespan.py

@@ -1,7 +1,7 @@
 import trio
 
 from ..config import Config
-from ..typing import ASGIFramework, LifespanScope
+from ..typing import ASGIFramework, ASGIReceiveEvent, ASGISendEvent, LifespanScope
 from ..utils import invoke_asgi, LifespanFailure, LifespanTimeout
 
 
@@ -63,10 +63,10 @@ async def wait_for_shutdown(self) -> None:
         except trio.TooSlowError as error:
             raise LifespanTimeout("startup") from error
 
-    async def asgi_receive(self) -> dict:
+    async def asgi_receive(self) -> ASGIReceiveEvent:
         return await self.app_receive_channel.receive()
 
-    async def asgi_send(self, message: dict) -> None:
+    async def asgi_send(self, message: ASGISendEvent) -> None:
         if message["type"] == "lifespan.startup.complete":
             self.startup.set()
         elif message["type"] == "lifespan.shutdown.complete":