Define HTTPProtocol class

uvicorn/_types.py

@@ -270,12 +270,5 @@ async def __call__(self, receive: ASGIReceiveCallable, send: ASGISendCallable) -
 
 
 ASGI2Application = type[ASGI2Protocol]
-ASGI3Application = Callable[
-    [
-        Scope,
-        ASGIReceiveCallable,
-        ASGISendCallable,
-    ],
-    Awaitable[None],
-]
+ASGI3Application = Callable[[Scope, ASGIReceiveCallable, ASGISendCallable], Awaitable[None]]
 ASGIApplication = Union[ASGI2Application, ASGI3Application]

uvicorn/logging.py

@@ -96,13 +96,7 @@ def default(code: int) -> str:
 
     def formatMessage(self, record: logging.LogRecord) -> str:
         recordcopy = copy(record)
-        (
-            client_addr,
-            method,
-            full_path,
-            http_version,
-            status_code,
-        ) = recordcopy.args  # type: ignore[misc]
+        (client_addr, method, full_path, http_version, status_code) = recordcopy.args  # type: ignore[misc]
         status_code = self.get_status_code(int(status_code))  # type: ignore[arg-type]
         request_line = f"{method} {full_path} HTTP/{http_version}"
         if self.use_colors:

uvicorn/protocols/http/base.py

@@ -0,0 +1,82 @@
+from __future__ import annotations as _annotations
+
+import asyncio
+import logging
+from typing import Any
+
+from uvicorn._types import HTTPScope
+from uvicorn.config import Config
+from uvicorn.protocols.http.flow_control import FlowControl
+from uvicorn.server import ServerState
+
+
+class HTTPProtocol(asyncio.Protocol):
+    __slots__ = (
+        "config",
+        "app",
+        "loop",
+        "logger",
+        "access_logger",
+        "access_log",
+        "ws_protocol_class",
+        "root_path",
+        "limit_concurrency",
+        "app_state",
+        # Timeouts
+        "timeout_keep_alive_task",
+        "timeout_keep_alive",
+        # Global state
+        "server_state",
+        "connections",
+        "tasks",
+        # Per-connection state
+        "transport",
+        "flow",
+        "server",
+        "client",
+        # Per-request state
+        "scope",
+        "headers",
+    )
+
+    def __init__(
+        self,
+        config: Config,
+        server_state: ServerState,
+        app_state: dict[str, Any],
+        _loop: asyncio.AbstractEventLoop | None = None,
+    ) -> None:
+        if not config.loaded:
+            config.load()
+
+        self.config = config
+        self.app = config.loaded_app
+        self.loop = _loop or asyncio.get_event_loop()
+
+        self.logger = logging.getLogger("uvicorn.error")
+        self.access_logger = logging.getLogger("uvicorn.access")
+        self.access_log = self.access_logger.hasHandlers()
+
+        self.ws_protocol_class = config.ws_protocol_class
+        self.root_path = config.root_path
+        self.limit_concurrency = config.limit_concurrency
+        self.app_state = app_state
+
+        # Timeouts
+        self.timeout_keep_alive_task: asyncio.TimerHandle | None = None
+        self.timeout_keep_alive = config.timeout_keep_alive
+
+        # Global state
+        self.server_state = server_state
+        self.connections = server_state.connections
+        self.tasks = server_state.tasks
+
+        # Per-connection state
+        self.transport: asyncio.Transport = None  # type: ignore[assignment]
+        self.flow: FlowControl = None  # type: ignore[assignment]
+        self.server: tuple[str, int] | None = None
+        self.client: tuple[str, int] | None = None
+
+        # Per-request state
+        self.scope: HTTPScope = None  # type: ignore[assignment]
+        self.headers: list[tuple[bytes, bytes]] = None  # type: ignore[assignment]

uvicorn/protocols/http/h11_impl.py

@@ -3,7 +3,7 @@
 import asyncio
 import http
 import logging
-from typing import Any, Callable, Literal, cast
+from typing import Any, Callable, cast
 from urllib.parse import unquote
 
 import h11
@@ -20,6 +20,7 @@
 )
 from uvicorn.config import Config
 from uvicorn.logging import TRACE_LOG_LEVEL
+from uvicorn.protocols.http.base import HTTPProtocol
 from uvicorn.protocols.http.flow_control import CLOSE_HEADER, HIGH_WATER_LIMIT, FlowControl, service_unavailable
 from uvicorn.protocols.utils import get_client_addr, get_local_addr, get_path_with_query_string, get_remote_addr, is_ssl
 from uvicorn.server import ServerState
@@ -35,63 +36,32 @@ def _get_status_phrase(status_code: int) -> bytes:
 STATUS_PHRASES = {status_code: _get_status_phrase(status_code) for status_code in range(100, 600)}
 
 
-class H11Protocol(asyncio.Protocol):
+class H11Protocol(HTTPProtocol):
     def __init__(
         self,
         config: Config,
         server_state: ServerState,
         app_state: dict[str, Any],
         _loop: asyncio.AbstractEventLoop | None = None,
     ) -> None:
-        if not config.loaded:
-            config.load()
-
-        self.config = config
-        self.app = config.loaded_app
-        self.loop = _loop or asyncio.get_event_loop()
-        self.logger = logging.getLogger("uvicorn.error")
-        self.access_logger = logging.getLogger("uvicorn.access")
-        self.access_log = self.access_logger.hasHandlers()
+        super().__init__(config, server_state, app_state, _loop)
+
         self.conn = h11.Connection(
             h11.SERVER,
             config.h11_max_incomplete_event_size
             if config.h11_max_incomplete_event_size is not None
             else DEFAULT_MAX_INCOMPLETE_EVENT_SIZE,
         )
-        self.ws_protocol_class = config.ws_protocol_class
-        self.root_path = config.root_path
-        self.limit_concurrency = config.limit_concurrency
-        self.app_state = app_state
-
-        # Timeouts
-        self.timeout_keep_alive_task: asyncio.TimerHandle | None = None
-        self.timeout_keep_alive = config.timeout_keep_alive
-
-        # Shared server state
-        self.server_state = server_state
-        self.connections = server_state.connections
-        self.tasks = server_state.tasks
-
-        # Per-connection state
-        self.transport: asyncio.Transport = None  # type: ignore[assignment]
-        self.flow: FlowControl = None  # type: ignore[assignment]
-        self.server: tuple[str, int] | None = None
-        self.client: tuple[str, int] | None = None
-        self.scheme: Literal["http", "https"] | None = None
 
         # Per-request state
-        self.scope: HTTPScope = None  # type: ignore[assignment]
-        self.headers: list[tuple[bytes, bytes]] = None  # type: ignore[assignment]
         self.cycle: RequestResponseCycle = None  # type: ignore[assignment]
 
     # Protocol interface
-    def connection_made(  # type: ignore[override]
-        self, transport: asyncio.Transport
-    ) -> None:
+    def connection_made(self, transport: asyncio.BaseTransport) -> None:
         self.connections.add(self)
 
-        self.transport = transport
-        self.flow = FlowControl(transport)
+        self.transport = cast(asyncio.Transport, transport)
+        self.flow = FlowControl(self.transport)
         self.server = get_local_addr(transport)
         self.client = get_remote_addr(transport)
         self.scheme = "https" if is_ssl(transport) else "http"
@@ -204,7 +174,7 @@ def handle_events(self) -> None:
                     "http_version": event.http_version.decode("ascii"),
                     "server": self.server,
                     "client": self.client,
-                    "scheme": self.scheme,  # type: ignore[typeddict-item]
+                    "scheme": self.scheme,
                     "method": event.method.decode("ascii"),
                     "root_path": self.root_path,
                     "path": full_path,
@@ -534,10 +504,6 @@ async def receive(self) -> ASGIReceiveEvent:
         if self.disconnected or self.response_complete:
             return {"type": "http.disconnect"}
 
-        message: HTTPRequestEvent = {
-            "type": "http.request",
-            "body": self.body,
-            "more_body": self.more_body,
-        }
+        message: HTTPRequestEvent = {"type": "http.request", "body": self.body, "more_body": self.more_body}
         self.body = b""
         return message

uvicorn/protocols/http/httptools_impl.py

@@ -5,9 +5,8 @@
 import logging
 import re
 import urllib
-from asyncio.events import TimerHandle
 from collections import deque
-from typing import Any, Callable, Literal, cast
+from typing import Any, Callable, cast
 
 import httptools
 
@@ -21,6 +20,7 @@
 )
 from uvicorn.config import Config
 from uvicorn.logging import TRACE_LOG_LEVEL
+from uvicorn.protocols.http.base import HTTPProtocol
 from uvicorn.protocols.http.flow_control import CLOSE_HEADER, HIGH_WATER_LIMIT, FlowControl, service_unavailable
 from uvicorn.protocols.utils import get_client_addr, get_local_addr, get_path_with_query_string, get_remote_addr, is_ssl
 from uvicorn.server import ServerState
@@ -40,23 +40,15 @@ def _get_status_line(status_code: int) -> bytes:
 STATUS_LINE = {status_code: _get_status_line(status_code) for status_code in range(100, 600)}
 
 
-class HttpToolsProtocol(asyncio.Protocol):
+class HttpToolsProtocol(HTTPProtocol):
     def __init__(
         self,
         config: Config,
         server_state: ServerState,
         app_state: dict[str, Any],
         _loop: asyncio.AbstractEventLoop | None = None,
     ) -> None:
-        if not config.loaded:
-            config.load()
-
-        self.config = config
-        self.app = config.loaded_app
-        self.loop = _loop or asyncio.get_event_loop()
-        self.logger = logging.getLogger("uvicorn.error")
-        self.access_logger = logging.getLogger("uvicorn.access")
-        self.access_log = self.access_logger.hasHandlers()
+        super().__init__(config, server_state, app_state, _loop)
         self.parser = httptools.HttpRequestParser(self)
 
         try:
@@ -66,42 +58,19 @@ def __init__(
             # httptools < 0.6.3
             pass
 
-        self.ws_protocol_class = config.ws_protocol_class
-        self.root_path = config.root_path
-        self.limit_concurrency = config.limit_concurrency
-        self.app_state = app_state
-
-        # Timeouts
-        self.timeout_keep_alive_task: TimerHandle | None = None
-        self.timeout_keep_alive = config.timeout_keep_alive
-
-        # Global state
-        self.server_state = server_state
-        self.connections = server_state.connections
-        self.tasks = server_state.tasks
-
         # Per-connection state
-        self.transport: asyncio.Transport = None  # type: ignore[assignment]
-        self.flow: FlowControl = None  # type: ignore[assignment]
-        self.server: tuple[str, int] | None = None
-        self.client: tuple[str, int] | None = None
-        self.scheme: Literal["http", "https"] | None = None
         self.pipeline: deque[tuple[RequestResponseCycle, ASGI3Application]] = deque()
 
         # Per-request state
-        self.scope: HTTPScope = None  # type: ignore[assignment]
-        self.headers: list[tuple[bytes, bytes]] = None  # type: ignore[assignment]
         self.expect_100_continue = False
         self.cycle: RequestResponseCycle = None  # type: ignore[assignment]
 
     # Protocol interface
-    def connection_made(  # type: ignore[override]
-        self, transport: asyncio.Transport
-    ) -> None:
+    def connection_made(self, transport: asyncio.BaseTransport) -> None:
         self.connections.add(self)
 
-        self.transport = transport
-        self.flow = FlowControl(transport)
+        self.transport = cast(asyncio.Transport, transport)
+        self.flow = FlowControl(self.transport)
         self.server = get_local_addr(transport)
         self.client = get_remote_addr(transport)
         self.scheme = "https" if is_ssl(transport) else "http"
@@ -226,7 +195,7 @@ def on_message_begin(self) -> None:
             "http_version": "1.1",
             "server": self.server,
             "client": self.client,
-            "scheme": self.scheme,  # type: ignore[typeddict-item]
+            "scheme": self.scheme,
             "root_path": self.root_path,
             "headers": self.headers,
             "state": self.app_state.copy(),

uvicorn/protocols/utils.py

@@ -9,7 +9,7 @@
 class ClientDisconnected(OSError): ...
 
 
-def get_remote_addr(transport: asyncio.Transport) -> tuple[str, int] | None:
+def get_remote_addr(transport: asyncio.BaseTransport) -> tuple[str, int] | None:
     socket_info = transport.get_extra_info("socket")
     if socket_info is not None:
         try:
@@ -26,7 +26,7 @@ def get_remote_addr(transport: asyncio.Transport) -> tuple[str, int] | None:
     return None
 
 
-def get_local_addr(transport: asyncio.Transport) -> tuple[str, int] | None:
+def get_local_addr(transport: asyncio.BaseTransport) -> tuple[str, int] | None:
     socket_info = transport.get_extra_info("socket")
     if socket_info is not None:
         info = socket_info.getsockname()
@@ -38,7 +38,7 @@ def get_local_addr(transport: asyncio.Transport) -> tuple[str, int] | None:
     return None
 
 
-def is_ssl(transport: asyncio.Transport) -> bool:
+def is_ssl(transport: asyncio.BaseTransport) -> bool:
     return bool(transport.get_extra_info("sslcontext"))