chore: improve type hints (#2867)

Author: waketzheng

pyproject.toml

@@ -91,7 +91,7 @@ source_pkgs = ["starlette", "tests"]
 [tool.coverage.report]
 exclude_lines = [
     "pragma: no cover",
-    "if typing.TYPE_CHECKING:",
-    "@typing.overload",
+    "if TYPE_CHECKING:",
+    "@overload",
     "raise NotImplementedError",
 ]

starlette/_exception_handler.py

@@ -1,6 +1,6 @@
 from __future__ import annotations
 
-import typing
+from typing import Any
 
 from starlette._utils import is_async_callable
 from starlette.concurrency import run_in_threadpool
@@ -9,7 +9,7 @@
 from starlette.types import ASGIApp, ExceptionHandler, Message, Receive, Scope, Send
 from starlette.websockets import WebSocket
 
-ExceptionHandlers = dict[typing.Any, ExceptionHandler]
+ExceptionHandlers = dict[Any, ExceptionHandler]
 StatusHandlers = dict[int, ExceptionHandler]
 
 

starlette/_utils.py

@@ -3,8 +3,9 @@
 import functools
 import inspect
 import sys
-import typing
-from contextlib import contextmanager
+from collections.abc import Awaitable, Generator
+from contextlib import AbstractAsyncContextManager, contextmanager
+from typing import Any, Callable, Generic, Protocol, TypeVar, overload
 
 from starlette.types import Scope
 
@@ -20,58 +21,58 @@
     except ImportError:
         has_exceptiongroups = False
 
-T = typing.TypeVar("T")
-AwaitableCallable = typing.Callable[..., typing.Awaitable[T]]
+T = TypeVar("T")
+AwaitableCallable = Callable[..., Awaitable[T]]
 
 
-@typing.overload
+@overload
 def is_async_callable(obj: AwaitableCallable[T]) -> TypeGuard[AwaitableCallable[T]]: ...
 
 
-@typing.overload
-def is_async_callable(obj: typing.Any) -> TypeGuard[AwaitableCallable[typing.Any]]: ...
+@overload
+def is_async_callable(obj: Any) -> TypeGuard[AwaitableCallable[Any]]: ...
 
 
-def is_async_callable(obj: typing.Any) -> typing.Any:
+def is_async_callable(obj: Any) -> Any:
     while isinstance(obj, functools.partial):
         obj = obj.func
 
     return inspect.iscoroutinefunction(obj) or (callable(obj) and inspect.iscoroutinefunction(obj.__call__))
 
 
-T_co = typing.TypeVar("T_co", covariant=True)
+T_co = TypeVar("T_co", covariant=True)
 
 
-class AwaitableOrContextManager(typing.Awaitable[T_co], typing.AsyncContextManager[T_co], typing.Protocol[T_co]): ...
+class AwaitableOrContextManager(Awaitable[T_co], AbstractAsyncContextManager[T_co], Protocol[T_co]): ...
 
 
-class SupportsAsyncClose(typing.Protocol):
+class SupportsAsyncClose(Protocol):
     async def close(self) -> None: ...  # pragma: no cover
 
 
-SupportsAsyncCloseType = typing.TypeVar("SupportsAsyncCloseType", bound=SupportsAsyncClose, covariant=False)
+SupportsAsyncCloseType = TypeVar("SupportsAsyncCloseType", bound=SupportsAsyncClose, covariant=False)
 
 
-class AwaitableOrContextManagerWrapper(typing.Generic[SupportsAsyncCloseType]):
+class AwaitableOrContextManagerWrapper(Generic[SupportsAsyncCloseType]):
     __slots__ = ("aw", "entered")
 
-    def __init__(self, aw: typing.Awaitable[SupportsAsyncCloseType]) -> None:
+    def __init__(self, aw: Awaitable[SupportsAsyncCloseType]) -> None:
         self.aw = aw
 
-    def __await__(self) -> typing.Generator[typing.Any, None, SupportsAsyncCloseType]:
+    def __await__(self) -> Generator[Any, None, SupportsAsyncCloseType]:
         return self.aw.__await__()
 
     async def __aenter__(self) -> SupportsAsyncCloseType:
         self.entered = await self.aw
         return self.entered
 
-    async def __aexit__(self, *args: typing.Any) -> None | bool:
+    async def __aexit__(self, *args: Any) -> None | bool:
         await self.entered.close()
         return None
 
 
 @contextmanager
-def collapse_excgroups() -> typing.Generator[None, None, None]:
+def collapse_excgroups() -> Generator[None, None, None]:
     try:
         yield
     except BaseException as exc:

starlette/applications.py

@@ -1,8 +1,9 @@
 from __future__ import annotations
 
 import sys
-import typing
 import warnings
+from collections.abc import Awaitable, Mapping, Sequence
+from typing import Any, Callable, TypeVar
 
 if sys.version_info >= (3, 10):  # pragma: no cover
     from typing import ParamSpec
@@ -20,7 +21,7 @@
 from starlette.types import ASGIApp, ExceptionHandler, Lifespan, Receive, Scope, Send
 from starlette.websockets import WebSocket
 
-AppType = typing.TypeVar("AppType", bound="Starlette")
+AppType = TypeVar("AppType", bound="Starlette")
 P = ParamSpec("P")
 
 
@@ -30,11 +31,11 @@ class Starlette:
     def __init__(
         self: AppType,
         debug: bool = False,
-        routes: typing.Sequence[BaseRoute] | None = None,
-        middleware: typing.Sequence[Middleware] | None = None,
-        exception_handlers: typing.Mapping[typing.Any, ExceptionHandler] | None = None,
-        on_startup: typing.Sequence[typing.Callable[[], typing.Any]] | None = None,
-        on_shutdown: typing.Sequence[typing.Callable[[], typing.Any]] | None = None,
+        routes: Sequence[BaseRoute] | None = None,
+        middleware: Sequence[Middleware] | None = None,
+        exception_handlers: Mapping[Any, ExceptionHandler] | None = None,
+        on_startup: Sequence[Callable[[], Any]] | None = None,
+        on_shutdown: Sequence[Callable[[], Any]] | None = None,
         lifespan: Lifespan[AppType] | None = None,
     ) -> None:
         """Initializes the application.
@@ -79,7 +80,7 @@ def __init__(
     def build_middleware_stack(self) -> ASGIApp:
         debug = self.debug
         error_handler = None
-        exception_handlers: dict[typing.Any, typing.Callable[[Request, Exception], Response]] = {}
+        exception_handlers: dict[Any, Callable[[Request, Exception], Response]] = {}
 
         for key, value in self.exception_handlers.items():
             if key in (500, Exception):
@@ -102,7 +103,7 @@ def build_middleware_stack(self) -> ASGIApp:
     def routes(self) -> list[BaseRoute]:
         return self.router.routes
 
-    def url_path_for(self, name: str, /, **path_params: typing.Any) -> URLPath:
+    def url_path_for(self, name: str, /, **path_params: Any) -> URLPath:
         return self.router.url_path_for(name, **path_params)
 
     async def __call__(self, scope: Scope, receive: Receive, send: Send) -> None:
@@ -111,7 +112,7 @@ async def __call__(self, scope: Scope, receive: Receive, send: Send) -> None:
             self.middleware_stack = self.build_middleware_stack()
         await self.middleware_stack(scope, receive, send)
 
-    def on_event(self, event_type: str) -> typing.Callable:  # type: ignore[type-arg]
+    def on_event(self, event_type: str) -> Callable:  # type: ignore[type-arg]
         return self.router.on_event(event_type)  # pragma: no cover
 
     def mount(self, path: str, app: ASGIApp, name: str | None = None) -> None:
@@ -140,14 +141,14 @@ def add_exception_handler(
     def add_event_handler(
         self,
         event_type: str,
-        func: typing.Callable,  # type: ignore[type-arg]
+        func: Callable,  # type: ignore[type-arg]
     ) -> None:  # pragma: no cover
         self.router.add_event_handler(event_type, func)
 
     def add_route(
         self,
         path: str,
-        route: typing.Callable[[Request], typing.Awaitable[Response] | Response],
+        route: Callable[[Request], Awaitable[Response] | Response],
         methods: list[str] | None = None,
         name: str | None = None,
         include_in_schema: bool = True,
@@ -157,19 +158,19 @@ def add_route(
     def add_websocket_route(
         self,
         path: str,
-        route: typing.Callable[[WebSocket], typing.Awaitable[None]],
+        route: Callable[[WebSocket], Awaitable[None]],
         name: str | None = None,
     ) -> None:  # pragma: no cover
         self.router.add_websocket_route(path, route, name=name)
 
-    def exception_handler(self, exc_class_or_status_code: int | type[Exception]) -> typing.Callable:  # type: ignore[type-arg]
+    def exception_handler(self, exc_class_or_status_code: int | type[Exception]) -> Callable:  # type: ignore[type-arg]
         warnings.warn(
             "The `exception_handler` decorator is deprecated, and will be removed in version 1.0.0. "
             "Refer to https://www.starlette.io/exceptions/ for the recommended approach.",
             DeprecationWarning,
         )
 
-        def decorator(func: typing.Callable) -> typing.Callable:  # type: ignore[type-arg]
+        def decorator(func: Callable) -> Callable:  # type: ignore[type-arg]
             self.add_exception_handler(exc_class_or_status_code, func)
             return func
 
@@ -181,7 +182,7 @@ def route(
         methods: list[str] | None = None,
         name: str | None = None,
         include_in_schema: bool = True,
-    ) -> typing.Callable:  # type: ignore[type-arg]
+    ) -> Callable:  # type: ignore[type-arg]
         """
         We no longer document this decorator style API, and its usage is discouraged.
         Instead you should use the following approach:
@@ -195,7 +196,7 @@ def route(
             DeprecationWarning,
         )
 
-        def decorator(func: typing.Callable) -> typing.Callable:  # type: ignore[type-arg]
+        def decorator(func: Callable) -> Callable:  # type: ignore[type-arg]
             self.router.add_route(
                 path,
                 func,
@@ -207,7 +208,7 @@ def decorator(func: typing.Callable) -> typing.Callable:  # type: ignore[type-ar
 
         return decorator
 
-    def websocket_route(self, path: str, name: str | None = None) -> typing.Callable:  # type: ignore[type-arg]
+    def websocket_route(self, path: str, name: str | None = None) -> Callable:  # type: ignore[type-arg]
         """
         We no longer document this decorator style API, and its usage is discouraged.
         Instead you should use the following approach:
@@ -221,13 +222,13 @@ def websocket_route(self, path: str, name: str | None = None) -> typing.Callable
             DeprecationWarning,
         )
 
-        def decorator(func: typing.Callable) -> typing.Callable:  # type: ignore[type-arg]
+        def decorator(func: Callable) -> Callable:  # type: ignore[type-arg]
             self.router.add_websocket_route(path, func, name=name)
             return func
 
         return decorator
 
-    def middleware(self, middleware_type: str) -> typing.Callable:  # type: ignore[type-arg]
+    def middleware(self, middleware_type: str) -> Callable:  # type: ignore[type-arg]
         """
         We no longer document this decorator style API, and its usage is discouraged.
         Instead you should use the following approach:
@@ -242,7 +243,7 @@ def middleware(self, middleware_type: str) -> typing.Callable:  # type: ignore[t
         )
         assert middleware_type == "http", 'Currently only middleware("http") is supported.'
 
-        def decorator(func: typing.Callable) -> typing.Callable:  # type: ignore[type-arg]
+        def decorator(func: Callable) -> Callable:  # type: ignore[type-arg]
             self.add_middleware(BaseHTTPMiddleware, dispatch=func)
             return func
 

starlette/authentication.py

@@ -3,7 +3,8 @@
 import functools
 import inspect
 import sys
-import typing
+from collections.abc import Sequence
+from typing import Any, Callable
 from urllib.parse import urlencode
 
 if sys.version_info >= (3, 10):  # pragma: no cover
@@ -20,23 +21,23 @@
 _P = ParamSpec("_P")
 
 
-def has_required_scope(conn: HTTPConnection, scopes: typing.Sequence[str]) -> bool:
+def has_required_scope(conn: HTTPConnection, scopes: Sequence[str]) -> bool:
     for scope in scopes:
         if scope not in conn.auth.scopes:
             return False
     return True
 
 
 def requires(
-    scopes: str | typing.Sequence[str],
+    scopes: str | Sequence[str],
     status_code: int = 403,
     redirect: str | None = None,
-) -> typing.Callable[[typing.Callable[_P, typing.Any]], typing.Callable[_P, typing.Any]]:
+) -> Callable[[Callable[_P, Any]], Callable[_P, Any]]:
     scopes_list = [scopes] if isinstance(scopes, str) else list(scopes)
 
     def decorator(
-        func: typing.Callable[_P, typing.Any],
-    ) -> typing.Callable[_P, typing.Any]:
+        func: Callable[_P, Any],
+    ) -> Callable[_P, Any]:
         sig = inspect.signature(func)
         for idx, parameter in enumerate(sig.parameters.values()):
             if parameter.name == "request" or parameter.name == "websocket":
@@ -62,7 +63,7 @@ async def websocket_wrapper(*args: _P.args, **kwargs: _P.kwargs) -> None:
         elif is_async_callable(func):
             # Handle async request/response functions.
             @functools.wraps(func)
-            async def async_wrapper(*args: _P.args, **kwargs: _P.kwargs) -> typing.Any:
+            async def async_wrapper(*args: _P.args, **kwargs: _P.kwargs) -> Any:
                 request = kwargs.get("request", args[idx] if idx < len(args) else None)
                 assert isinstance(request, Request)
 
@@ -79,7 +80,7 @@ async def async_wrapper(*args: _P.args, **kwargs: _P.kwargs) -> typing.Any:
         else:
             # Handle sync request/response functions.
             @functools.wraps(func)
-            def sync_wrapper(*args: _P.args, **kwargs: _P.kwargs) -> typing.Any:
+            def sync_wrapper(*args: _P.args, **kwargs: _P.kwargs) -> Any:
                 request = kwargs.get("request", args[idx] if idx < len(args) else None)
                 assert isinstance(request, Request)
 
@@ -106,7 +107,7 @@ async def authenticate(self, conn: HTTPConnection) -> tuple[AuthCredentials, Bas
 
 
 class AuthCredentials:
-    def __init__(self, scopes: typing.Sequence[str] | None = None):
+    def __init__(self, scopes: Sequence[str] | None = None):
         self.scopes = [] if scopes is None else list(scopes)
 
 

starlette/background.py

@@ -1,7 +1,8 @@
 from __future__ import annotations
 
 import sys
-import typing
+from collections.abc import Sequence
+from typing import Any, Callable
 
 if sys.version_info >= (3, 10):  # pragma: no cover
     from typing import ParamSpec
@@ -15,7 +16,7 @@
 
 
 class BackgroundTask:
-    def __init__(self, func: typing.Callable[P, typing.Any], *args: P.args, **kwargs: P.kwargs) -> None:
+    def __init__(self, func: Callable[P, Any], *args: P.args, **kwargs: P.kwargs) -> None:
         self.func = func
         self.args = args
         self.kwargs = kwargs
@@ -29,10 +30,10 @@ async def __call__(self) -> None:
 
 
 class BackgroundTasks(BackgroundTask):
-    def __init__(self, tasks: typing.Sequence[BackgroundTask] | None = None):
+    def __init__(self, tasks: Sequence[BackgroundTask] | None = None):
         self.tasks = list(tasks) if tasks else []
 
-    def add_task(self, func: typing.Callable[P, typing.Any], *args: P.args, **kwargs: P.kwargs) -> None:
+    def add_task(self, func: Callable[P, Any], *args: P.args, **kwargs: P.kwargs) -> None:
         task = BackgroundTask(func, *args, **kwargs)
         self.tasks.append(task)