Follow spec

a2wsgi/asgi.py

@@ -3,11 +3,11 @@
 import threading
 from http import HTTPStatus
 from io import BytesIO
-from itertools import chain
 from typing import Any, Coroutine, Deque, Iterable, Optional
 from typing import cast as typing_cast
 
-from .types import ASGIApp, Environ, ExcInfo, Message, Scope, StartResponse
+from .asgi_typing import HTTPScope, ASGIApp, ReceiveEvent, SendEvent
+from .wsgi_typing import Environ, StartResponse, ExceptionInfo, IterableChunks
 
 
 class defaultdict(dict):
@@ -70,46 +70,45 @@ def wait(self) -> Any:
         return message
 
 
-def build_scope(environ: Environ) -> Scope:
+def build_scope(environ: Environ) -> HTTPScope:
     headers = [
-        (key.lower().replace("_", "-").encode("latin-1"), value.encode("latin-1"))
-        for key, value in chain(
-            (
-                (key[5:], value)
-                for key, value in environ.items()
-                if key.startswith("HTTP_")
-                and key not in ("HTTP_CONTENT_TYPE", "HTTP_CONTENT_LENGTH")
-            ),
-            (
-                (key, value)
-                for key, value in environ.items()
-                if key in ("CONTENT_TYPE", "CONTENT_LENGTH")
-            ),
+        (
+            (key[5:] if key.startswith("HTTP_") else key)
+            .lower()
+            .replace("_", "-")
+            .encode("latin-1"),
+            value.encode("latin-1"),  # type: ignore
         )
+        for key, value in environ.items()
+        if (
+            key.startswith("HTTP_")
+            and key not in ("HTTP_CONTENT_TYPE", "HTTP_CONTENT_LENGTH")
+        )
+        or key in ("CONTENT_TYPE", "CONTENT_LENGTH")
     ]
 
-    if environ.get("REMOTE_ADDR") and environ.get("REMOTE_PORT"):
-        client = (environ["REMOTE_ADDR"], int(environ["REMOTE_PORT"]))
-    else:
-        client = None
-
     root_path = environ.get("SCRIPT_NAME", "").encode("latin1").decode("utf8")
-    path = root_path + environ["PATH_INFO"].encode("latin1").decode("utf8")
+    path = root_path + environ.get("PATH_INFO", "").encode("latin1").decode("utf8")
 
-    return {
-        "wsgi_environ": environ,
+    scope: HTTPScope = {
+        "wsgi_environ": environ,  # type: ignore a2wsgi
         "type": "http",
         "asgi": {"version": "3.0", "spec_version": "3.0"},
         "http_version": environ.get("SERVER_PROTOCOL", "http/1.0").split("/")[1],
         "method": environ["REQUEST_METHOD"],
         "scheme": environ.get("wsgi.url_scheme", "http"),
         "path": path,
-        "query_string": environ["QUERY_STRING"].encode("ascii"),
+        "query_string": environ.get("QUERY_STRING", "").encode("ascii"),
         "root_path": root_path,
-        "client": client,
         "server": (environ["SERVER_NAME"], int(environ["SERVER_PORT"])),
         "headers": headers,
+        "extensions": {},
     }
+    if environ.get("REMOTE_ADDR") and environ.get("REMOTE_PORT"):
+        client = (environ.get("REMOTE_ADDR", ""), int(environ.get("REMOTE_PORT", "0")))
+        scope["client"] = client
+
+    return scope
 
 
 class ASGIMiddleware:
@@ -164,12 +163,12 @@ def _init_async_lock():
         self.asgi_done = threading.Event()
         self.wsgi_should_stop: bool = False
 
-    async def asgi_receive(self) -> Message:
+    async def asgi_receive(self) -> ReceiveEvent:
         async with self.async_lock:
             self.sync_event.set({"type": "receive"})
             return await self.async_event.wait()
 
-    async def asgi_send(self, message: Message) -> None:
+    async def asgi_send(self, message: SendEvent) -> None:
         async with self.async_lock:
             self.sync_event.set(message)
             await self.async_event.wait()
@@ -201,7 +200,7 @@ def start_asgi_app(self, environ: Environ) -> asyncio.Task:
 
     def __call__(
         self, environ: Environ, start_response: StartResponse
-    ) -> Iterable[bytes]:
+    ) -> IterableChunks:
         read_count: int = 0
         body = environ["wsgi.input"] or BytesIO()
         content_length = int(environ.get("CONTENT_LENGTH", None) or 0)
@@ -262,8 +261,8 @@ def __call__(
         yield b""
 
     def error_response(
-        self, start_response: StartResponse, exception: ExcInfo
-    ) -> Iterable[bytes]:
+        self, start_response: StartResponse, exception: ExceptionInfo
+    ) -> IterableChunks:
         start_response(
             "500 Internal Server Error",
             [

a2wsgi/asgi_typing.py

@@ -0,0 +1,182 @@
+"""
+https://asgi.readthedocs.io/en/latest/specs/index.html
+"""
+import sys
+from typing import (
+    Any,
+    Awaitable,
+    Callable,
+    Dict,
+    Iterable,
+    Literal,
+    Optional,
+    Tuple,
+    TypedDict,
+    Union,
+)
+
+if sys.version_info >= (3, 11):
+    from typing import NotRequired
+else:
+    from typing_extensions import NotRequired
+
+
+class ASGIVersions(TypedDict):
+    spec_version: str
+    version: Literal["3.0"]
+
+
+class HTTPScope(TypedDict):
+    type: Literal["http"]
+    asgi: ASGIVersions
+    http_version: str
+    method: str
+    scheme: str
+    path: str
+    raw_path: NotRequired[bytes]
+    query_string: bytes
+    root_path: str
+    headers: Iterable[Tuple[bytes, bytes]]
+    client: NotRequired[Tuple[str, int]]
+    server: NotRequired[Tuple[str, Optional[int]]]
+    state: NotRequired[Dict[str, Any]]
+    extensions: NotRequired[Dict[str, Dict[object, object]]]
+
+
+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: NotRequired[Tuple[str, int]]
+    server: NotRequired[Tuple[str, Optional[int]]]
+    subprotocols: Iterable[str]
+    state: NotRequired[Dict[str, Any]]
+    extensions: NotRequired[Dict[str, Dict[object, object]]]
+
+
+class LifespanScope(TypedDict):
+    type: Literal["lifespan"]
+    asgi: ASGIVersions
+    state: NotRequired[Dict[str, Any]]
+
+
+WWWScope = Union[HTTPScope, WebSocketScope]
+Scope = Union[HTTPScope, WebSocketScope, LifespanScope]
+
+
+class HTTPRequestEvent(TypedDict):
+    type: Literal["http.request"]
+    body: bytes
+    more_body: NotRequired[bool]
+
+
+class HTTPResponseStartEvent(TypedDict):
+    type: Literal["http.response.start"]
+    status: int
+    headers: NotRequired[Iterable[Tuple[bytes, bytes]]]
+    trailers: NotRequired[bool]
+
+
+class HTTPResponseBodyEvent(TypedDict):
+    type: Literal["http.response.body"]
+    body: NotRequired[bytes]
+    more_body: NotRequired[bool]
+
+
+class HTTPDisconnectEvent(TypedDict):
+    type: Literal["http.disconnect"]
+
+
+class WebSocketConnectEvent(TypedDict):
+    type: Literal["websocket.connect"]
+
+
+class WebSocketAcceptEvent(TypedDict):
+    type: Literal["websocket.accept"]
+    subprotocol: NotRequired[str]
+    headers: NotRequired[Iterable[Tuple[bytes, bytes]]]
+
+
+class WebSocketReceiveEvent(TypedDict):
+    type: Literal["websocket.receive"]
+    bytes: NotRequired[bytes]
+    text: NotRequired[str]
+
+
+class WebSocketSendEvent(TypedDict):
+    type: Literal["websocket.send"]
+    bytes: NotRequired[bytes]
+    text: NotRequired[str]
+
+
+class WebSocketDisconnectEvent(TypedDict):
+    type: Literal["websocket.disconnect"]
+    code: int
+
+
+class WebSocketCloseEvent(TypedDict):
+    type: Literal["websocket.close"]
+    code: NotRequired[int]
+    reason: NotRequired[str]
+
+
+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
+
+
+ReceiveEvent = Union[
+    HTTPRequestEvent,
+    HTTPDisconnectEvent,
+    WebSocketConnectEvent,
+    WebSocketReceiveEvent,
+    WebSocketDisconnectEvent,
+    LifespanStartupEvent,
+    LifespanShutdownEvent,
+]
+
+SendEvent = Union[
+    HTTPResponseStartEvent,
+    HTTPResponseBodyEvent,
+    HTTPDisconnectEvent,
+    WebSocketAcceptEvent,
+    WebSocketSendEvent,
+    WebSocketCloseEvent,
+    LifespanStartupCompleteEvent,
+    LifespanStartupFailedEvent,
+    LifespanShutdownCompleteEvent,
+    LifespanShutdownFailedEvent,
+]
+
+Receive = Callable[[], Awaitable[ReceiveEvent]]
+
+Send = Callable[[SendEvent], Awaitable[None]]
+
+ASGIApp = Callable[[Scope, Receive, Send], Awaitable[None]]