chore: improve type hints (#2638)

Author: waketzheng

tests/middleware/test_logging.py

@@ -4,7 +4,8 @@
 import logging
 import socket
 import sys
-import typing
+from collections.abc import Iterator
+from typing import TYPE_CHECKING
 
 import httpx
 import pytest
@@ -15,7 +16,7 @@
 from uvicorn import Config
 from uvicorn._types import ASGIReceiveCallable, ASGISendCallable, Scope
 
-if typing.TYPE_CHECKING:
+if TYPE_CHECKING:
     import sys
 
     from uvicorn.protocols.websockets.websockets_impl import WebSocketProtocol
@@ -32,7 +33,7 @@
 
 
 @contextlib.contextmanager
-def caplog_for_logger(caplog: pytest.LogCaptureFixture, logger_name: str) -> typing.Iterator[pytest.LogCaptureFixture]:
+def caplog_for_logger(caplog: pytest.LogCaptureFixture, logger_name: str) -> Iterator[pytest.LogCaptureFixture]:
     logger = logging.getLogger(logger_name)
     logger.propagate, old_propagate = False, logger.propagate
     logger.addHandler(caplog.handler)

tests/protocols/test_websocket.py

@@ -1,8 +1,8 @@
 from __future__ import annotations
 
 import asyncio
-import typing
 from copy import deepcopy
+from typing import TYPE_CHECKING, Any, TypedDict
 
 import httpx
 import pytest
@@ -35,7 +35,7 @@
 except ModuleNotFoundError:  # pragma: no cover
     skip_if_no_wsproto = pytest.mark.skipif(True, reason="wsproto is not installed.")
 
-if typing.TYPE_CHECKING:
+if TYPE_CHECKING:
     import sys
 
     from uvicorn.protocols.http.h11_impl import H11Protocol
@@ -776,7 +776,7 @@ async def websocket_session(url: str):
     assert disconnected_message == {"type": "websocket.disconnect", "code": 1006}
 
 
-class EmptyDict(typing.TypedDict): ...
+class EmptyDict(TypedDict): ...
 
 
 async def test_server_reject_connection_with_response(
@@ -1142,12 +1142,12 @@ async def open_connection(url: str):
 
 
 async def test_lifespan_state(ws_protocol_cls: WSProtocol, http_protocol_cls: HTTPProtocol, unused_tcp_port: int):
-    expected_states: list[dict[str, typing.Any]] = [
+    expected_states: list[dict[str, Any]] = [
         {"a": 123, "b": [1]},
         {"a": 123, "b": [1, 2]},
     ]
 
-    actual_states: list[dict[str, typing.Any]] = []
+    actual_states: list[dict[str, Any]] = []
 
     async def lifespan_app(scope: Scope, receive: ASGIReceiveCallable, send: ASGISendCallable):
         message = await receive()

tests/test_config.py

@@ -7,9 +7,9 @@
 import os
 import socket
 import sys
-import typing
+from collections.abc import Iterator
 from pathlib import Path
-from typing import Any, Literal
+from typing import IO, Any, Callable, Literal
 from unittest.mock import MagicMock
 
 import pytest
@@ -291,7 +291,7 @@ def test_ssl_config_combined(tls_certificate_key_and_chain_path: str) -> None:
     assert config.is_ssl is True
 
 
-def asgi2_app(scope: Scope) -> typing.Callable:
+def asgi2_app(scope: Scope) -> Callable:
     async def asgi(receive: ASGIReceiveCallable, send: ASGISendCallable) -> None:  # pragma: nocover
         pass
 
@@ -374,7 +374,7 @@ def test_log_config_yaml(
 @pytest.mark.parametrize("config_file", ["log_config.ini", configparser.ConfigParser(), io.StringIO()])
 def test_log_config_file(
     mocked_logging_config_module: MagicMock,
-    config_file: str | configparser.RawConfigParser | typing.IO[Any],
+    config_file: str | configparser.RawConfigParser | IO[Any],
 ) -> None:
     """
     Test that one can load a configparser config from disk.
@@ -386,14 +386,14 @@ def test_log_config_file(
 
 
 @pytest.fixture(params=[0, 1])
-def web_concurrency(request: pytest.FixtureRequest) -> typing.Iterator[int]:
+def web_concurrency(request: pytest.FixtureRequest) -> Iterator[int]:
     yield request.param
     if os.getenv("WEB_CONCURRENCY"):
         del os.environ["WEB_CONCURRENCY"]
 
 
 @pytest.fixture(params=["127.0.0.1", "127.0.0.2"])
-def forwarded_allow_ips(request: pytest.FixtureRequest) -> typing.Iterator[str]:
+def forwarded_allow_ips(request: pytest.FixtureRequest) -> Iterator[str]:
     yield request.param
     if os.getenv("FORWARDED_ALLOW_IPS"):
         del os.environ["FORWARDED_ALLOW_IPS"]

uvicorn/protocols/websockets/auto.py

@@ -1,9 +1,9 @@
 from __future__ import annotations
 
 import asyncio
-import typing
+from typing import Callable
 
-AutoWebSocketsProtocol: typing.Callable[..., asyncio.Protocol] | None
+AutoWebSocketsProtocol: Callable[..., asyncio.Protocol] | None
 try:
     import websockets  # noqa
 except ImportError:  # pragma: no cover

uvicorn/protocols/websockets/wsproto_impl.py

@@ -2,8 +2,7 @@
 
 import asyncio
 import logging
-import typing
-from typing import Literal, cast
+from typing import Any, Literal, cast
 from urllib.parse import unquote
 
 import wsproto
@@ -41,7 +40,7 @@ def __init__(
         self,
         config: Config,
         server_state: ServerState,
-        app_state: dict[str, typing.Any],
+        app_state: dict[str, Any],
         _loop: asyncio.AbstractEventLoop | None = None,
     ) -> None:
         if not config.loaded:
@@ -256,7 +255,7 @@ async def send(self, message: ASGISendEvent) -> None:
 
         if not self.handshake_complete:
             if message_type == "websocket.accept":
-                message = typing.cast(WebSocketAcceptEvent, message)
+                message = cast(WebSocketAcceptEvent, message)
                 self.logger.info(
                     '%s - "WebSocket %s" [accepted]',
                     get_client_addr(self.scope),
@@ -293,7 +292,7 @@ async def send(self, message: ASGISendEvent) -> None:
                 self.transport.close()
 
             elif message_type == "websocket.http.response.start":
-                message = typing.cast(WebSocketResponseStartEvent, message)
+                message = cast(WebSocketResponseStartEvent, message)
                 # ensure status code is in the valid range
                 if not (100 <= message["status"] < 600):
                     msg = "Invalid HTTP status code '%d' in response."
@@ -325,7 +324,7 @@ async def send(self, message: ASGISendEvent) -> None:
         elif not self.close_sent and not self.response_started:
             try:
                 if message_type == "websocket.send":
-                    message = typing.cast(WebSocketSendEvent, message)
+                    message = cast(WebSocketSendEvent, message)
                     bytes_data = message.get("bytes")
                     text_data = message.get("text")
                     data = text_data if bytes_data is None else bytes_data
@@ -334,7 +333,7 @@ async def send(self, message: ASGISendEvent) -> None:
                         self.transport.write(output)
 
                 elif message_type == "websocket.close":
-                    message = typing.cast(WebSocketCloseEvent, message)
+                    message = cast(WebSocketCloseEvent, message)
                     self.close_sent = True
                     code = message.get("code", 1000)
                     reason = message.get("reason", "") or ""
@@ -351,7 +350,7 @@ async def send(self, message: ASGISendEvent) -> None:
                 raise ClientDisconnected from exc
         elif self.response_started:
             if message_type == "websocket.http.response.body":
-                message = typing.cast("WebSocketResponseBodyEvent", message)
+                message = cast("WebSocketResponseBodyEvent", message)
                 body_finished = not message.get("more_body", False)
                 reject_data = events.RejectData(data=message["body"], body_finished=body_finished)
                 output = self.conn.send(reject_data)