chore: typing cleanup (#2339)

* refactor: privatize typing module

* refactor: rename missing to unset

* typing: define public typing module

Other minor changes in the header types

* docs: improved attributes docs by inlining them

* typing: improve secure_filename typing

* docs: fix linter issues

* docs: make sphinx happy

* docs: improve attribute docs

* typing: finalize typing module

* docs: workaround sphinx autoclass typing rendering bug

* docs: add typing aliases to the docs

also customize how generic aliases docs are rendered to avoid
adding "alias of <the real type>"

* chore: update .coveragerc typing => _typing

* chore: update after review

* fix: fix typo

---------

Co-authored-by: Vytautas Liuolia <vytautas.liuolia@gmail.com>

.coveragerc

@@ -1,7 +1,7 @@
 [run]
 branch = True
 source = falcon
-omit = falcon/tests*,falcon/typing.py,falcon/cmd/bench.py,falcon/bench/*,falcon/vendor/*
+omit = falcon/tests*,falcon/_typing.py,falcon/cmd/bench.py,falcon/bench/*,falcon/vendor/*
 
 parallel = True
 

docs/api/util.rst

@@ -85,3 +85,9 @@ Other
 
 .. autoclass:: falcon.ETag
     :members:
+
+Type Aliases
+------------
+
+.. automodule:: falcon.typing
+    :members:

docs/conf.py

@@ -77,6 +77,7 @@
     'sphinx_design',
     'myst_parser',
     # Falcon-specific extensions
+    'ext.autodoc_customizations',
     'ext.cibuildwheel',
     'ext.doorway',
     'ext.private_args',

docs/ext/autodoc_customizations.py

@@ -0,0 +1,10 @@
+"""Customizations to the autodoc functionalities"""
+
+import sphinx.ext.autodoc as ad
+
+
+def setup(app):
+    # avoid adding "alias of xyz"
+    ad.GenericAliasMixin.update_content = ad.DataDocumenterMixinBase.update_content
+
+    return {'parallel_read_safe': True}

falcon/_typing.py

@@ -0,0 +1,198 @@
+# Copyright 2021-2023 by Vytautas Liuolia.
+#
+# Licensed under the Apache License, Version 2.0 (the "License");
+# you may not use this file except in compliance with the License.
+# You may obtain a copy of the License at
+#
+#    http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+"""Private type aliases used internally by Falcon.."""
+
+from __future__ import annotations
+
+from enum import auto
+from enum import Enum
+import http
+from http.cookiejar import Cookie
+import sys
+from typing import (
+    Any,
+    Awaitable,
+    Callable,
+    Dict,
+    Iterable,
+    List,
+    Literal,
+    Mapping,
+    Optional,
+    Pattern,
+    Protocol,
+    Tuple,
+    TYPE_CHECKING,
+    TypeVar,
+    Union,
+)
+
+# NOTE(vytas): Mypy still struggles to handle a conditional import in the EAFP
+#   fashion, so we branch on Py version instead (which it does understand).
+if sys.version_info >= (3, 11):
+    from wsgiref.types import StartResponse as StartResponse
+    from wsgiref.types import WSGIEnvironment as WSGIEnvironment
+else:
+    WSGIEnvironment = Dict[str, Any]
+    StartResponse = Callable[[str, List[Tuple[str, str]]], Callable[[bytes], None]]
+
+if TYPE_CHECKING:
+    from falcon.asgi import Request as AsgiRequest
+    from falcon.asgi import Response as AsgiResponse
+    from falcon.asgi import WebSocket
+    from falcon.asgi_spec import AsgiEvent
+    from falcon.asgi_spec import AsgiSendMsg
+    from falcon.http_error import HTTPError
+    from falcon.request import Request
+    from falcon.response import Response
+
+
+class _Unset(Enum):
+    UNSET = auto()
+
+
+_T = TypeVar('_T')
+_UNSET = _Unset.UNSET
+UnsetOr = Union[Literal[_Unset.UNSET], _T]
+
+Link = Dict[str, str]
+CookieArg = Mapping[str, Union[str, Cookie]]
+# Error handlers
+ErrorHandler = Callable[['Request', 'Response', BaseException, Dict[str, Any]], None]
+
+
+class AsgiErrorHandler(Protocol):
+    async def __call__(
+        self,
+        req: AsgiRequest,
+        resp: Optional[AsgiResponse],
+        error: BaseException,
+        params: Dict[str, Any],
+        *,
+        ws: Optional[WebSocket] = ...,
+    ) -> None: ...
+
+
+# Error serializers
+ErrorSerializer = Callable[['Request', 'Response', 'HTTPError'], None]
+
+# Sinks
+SinkPrefix = Union[str, Pattern[str]]
+
+
+class SinkCallable(Protocol):
+    def __call__(self, req: Request, resp: Response, **kwargs: str) -> None: ...
+
+
+class AsgiSinkCallable(Protocol):
+    async def __call__(
+        self, req: AsgiRequest, resp: AsgiResponse, **kwargs: str
+    ) -> None: ...
+
+
+HeaderMapping = Mapping[str, str]
+HeaderIter = Iterable[Tuple[str, str]]
+HeaderArg = Union[HeaderMapping, HeaderIter]
+ResponseStatus = Union[http.HTTPStatus, str, int]
+StoreArg = Optional[Dict[str, Any]]
+Resource = object
+RangeSetHeader = Union[Tuple[int, int, int], Tuple[int, int, int, str]]
+
+
+# WSGI
+class ResponderMethod(Protocol):
+    def __call__(
+        self,
+        resource: Resource,
+        req: Request,
+        resp: Response,
+        **kwargs: Any,
+    ) -> None: ...
+
+
+class ResponderCallable(Protocol):
+    def __call__(self, req: Request, resp: Response, **kwargs: Any) -> None: ...
+
+
+ProcessRequestMethod = Callable[['Request', 'Response'], None]
+ProcessResourceMethod = Callable[
+    ['Request', 'Response', Resource, Dict[str, Any]], None
+]
+ProcessResponseMethod = Callable[['Request', 'Response', Resource, bool], None]
+
+
+# ASGI
+class AsgiResponderMethod(Protocol):
+    async def __call__(
+        self,
+        resource: Resource,
+        req: AsgiRequest,
+        resp: AsgiResponse,
+        **kwargs: Any,
+    ) -> None: ...
+
+
+class AsgiResponderCallable(Protocol):
+    async def __call__(
+        self, req: AsgiRequest, resp: AsgiResponse, **kwargs: Any
+    ) -> None: ...
+
+
+class AsgiResponderWsCallable(Protocol):
+    async def __call__(
+        self, req: AsgiRequest, ws: WebSocket, **kwargs: Any
+    ) -> None: ...
+
+
+AsgiReceive = Callable[[], Awaitable['AsgiEvent']]
+AsgiSend = Callable[['AsgiSendMsg'], Awaitable[None]]
+AsgiProcessRequestMethod = Callable[['AsgiRequest', 'AsgiResponse'], Awaitable[None]]
+AsgiProcessResourceMethod = Callable[
+    ['AsgiRequest', 'AsgiResponse', Resource, Dict[str, Any]], Awaitable[None]
+]
+AsgiProcessResponseMethod = Callable[
+    ['AsgiRequest', 'AsgiResponse', Resource, bool], Awaitable[None]
+]
+AsgiProcessRequestWsMethod = Callable[['AsgiRequest', 'WebSocket'], Awaitable[None]]
+AsgiProcessResourceWsMethod = Callable[
+    ['AsgiRequest', 'WebSocket', Resource, Dict[str, Any]], Awaitable[None]
+]
+ResponseCallbacks = Union[
+    Tuple[Callable[[], None], Literal[False]],
+    Tuple[Callable[[], Awaitable[None]], Literal[True]],
+]
+
+
+# Routing
+
+MethodDict = Union[
+    Dict[str, ResponderCallable],
+    Dict[str, Union[AsgiResponderCallable, AsgiResponderWsCallable]],
+]
+
+
+class FindMethod(Protocol):
+    def __call__(
+        self, uri: str, req: Optional[Request]
+    ) -> Optional[Tuple[object, MethodDict, Dict[str, Any], Optional[str]]]: ...
+
+
+# Media
+class SerializeSync(Protocol):
+    def __call__(self, media: Any, content_type: Optional[str] = ...) -> bytes: ...
+
+
+DeserializeSync = Callable[[bytes], Any]
+
+Responder = Union[ResponderMethod, AsgiResponderMethod]

falcon/app.py

@@ -45,6 +45,18 @@
 from falcon import constants
 from falcon import responders
 from falcon import routing
+from falcon._typing import AsgiResponderCallable
+from falcon._typing import AsgiResponderWsCallable
+from falcon._typing import AsgiSinkCallable
+from falcon._typing import ErrorHandler
+from falcon._typing import ErrorSerializer
+from falcon._typing import FindMethod
+from falcon._typing import ProcessResponseMethod
+from falcon._typing import ResponderCallable
+from falcon._typing import SinkCallable
+from falcon._typing import SinkPrefix
+from falcon._typing import StartResponse
+from falcon._typing import WSGIEnvironment
 from falcon.errors import CompatibilityError
 from falcon.errors import HTTPBadRequest
 from falcon.errors import HTTPInternalServerError
@@ -56,19 +68,7 @@
 from falcon.response import Response
 from falcon.response import ResponseOptions
 import falcon.status_codes as status
-from falcon.typing import AsgiResponderCallable
-from falcon.typing import AsgiResponderWsCallable
-from falcon.typing import AsgiSinkCallable
-from falcon.typing import ErrorHandler
-from falcon.typing import ErrorSerializer
-from falcon.typing import FindMethod
-from falcon.typing import ProcessResponseMethod
 from falcon.typing import ReadableIO
-from falcon.typing import ResponderCallable
-from falcon.typing import SinkCallable
-from falcon.typing import SinkPrefix
-from falcon.typing import StartResponse
-from falcon.typing import WSGIEnvironment
 from falcon.util import deprecation
 from falcon.util import misc
 from falcon.util.misc import code_to_http_status
@@ -303,8 +303,8 @@ def process_response(
     def __init__(
         self,
         media_type: str = constants.DEFAULT_MEDIA_TYPE,
-        request_type: Type[Request] = Request,
-        response_type: Type[Response] = Response,
+        request_type: Optional[Type[Request]] = None,
+        response_type: Optional[Type[Response]] = None,
         middleware: Union[object, Iterable[object]] = None,
         router: Optional[routing.CompiledRouter] = None,
         independent_middleware: bool = True,
@@ -342,8 +342,8 @@ def __init__(
         self._router = router or routing.DefaultRouter()
         self._router_search = self._router.find
 
-        self._request_type = request_type
-        self._response_type = response_type
+        self._request_type = request_type or Request
+        self._response_type = response_type or Response
 
         self._error_handlers = {}
         self._serialize_error = helpers.default_serialize_error

falcon/app_helpers.py

@@ -20,20 +20,20 @@
 from typing import IO, Iterable, List, Literal, Optional, overload, Tuple, Union
 
 from falcon import util
+from falcon._typing import AsgiProcessRequestMethod as APRequest
+from falcon._typing import AsgiProcessRequestWsMethod
+from falcon._typing import AsgiProcessResourceMethod as APResource
+from falcon._typing import AsgiProcessResourceWsMethod
+from falcon._typing import AsgiProcessResponseMethod as APResponse
+from falcon._typing import ProcessRequestMethod as PRequest
+from falcon._typing import ProcessResourceMethod as PResource
+from falcon._typing import ProcessResponseMethod as PResponse
 from falcon.constants import MEDIA_JSON
 from falcon.constants import MEDIA_XML
 from falcon.errors import CompatibilityError
 from falcon.errors import HTTPError
 from falcon.request import Request
 from falcon.response import Response
-from falcon.typing import AsgiProcessRequestMethod as APRequest
-from falcon.typing import AsgiProcessRequestWsMethod
-from falcon.typing import AsgiProcessResourceMethod as APResource
-from falcon.typing import AsgiProcessResourceWsMethod
-from falcon.typing import AsgiProcessResponseMethod as APResponse
-from falcon.typing import ProcessRequestMethod as PRequest
-from falcon.typing import ProcessResourceMethod as PResource
-from falcon.typing import ProcessResponseMethod as PResponse
 from falcon.util.sync import _wrap_non_coroutine_unsafe
 
 __all__ = (

falcon/asgi/app.py

@@ -40,6 +40,14 @@
 from falcon import constants
 from falcon import responders
 from falcon import routing
+from falcon._typing import _UNSET
+from falcon._typing import AsgiErrorHandler
+from falcon._typing import AsgiReceive
+from falcon._typing import AsgiResponderCallable
+from falcon._typing import AsgiResponderWsCallable
+from falcon._typing import AsgiSend
+from falcon._typing import AsgiSinkCallable
+from falcon._typing import SinkPrefix
 import falcon.app
 from falcon.app_helpers import AsyncPreparedMiddlewareResult
 from falcon.app_helpers import AsyncPreparedMiddlewareWsResult
@@ -55,14 +63,6 @@
 from falcon.http_error import HTTPError
 from falcon.http_status import HTTPStatus
 from falcon.media.multipart import MultipartFormHandler
-from falcon.typing import AsgiErrorHandler
-from falcon.typing import AsgiReceive
-from falcon.typing import AsgiResponderCallable
-from falcon.typing import AsgiResponderWsCallable
-from falcon.typing import AsgiSend
-from falcon.typing import AsgiSinkCallable
-from falcon.typing import MISSING
-from falcon.typing import SinkPrefix
 from falcon.util import get_argnames
 from falcon.util.misc import is_python_func
 from falcon.util.sync import _should_wrap_non_coroutines
@@ -366,8 +366,8 @@ async def process_resource_ws(
     def __init__(
         self,
         media_type: str = constants.DEFAULT_MEDIA_TYPE,
-        request_type: Type[Request] = Request,
-        response_type: Type[Response] = Response,
+        request_type: Optional[Type[Request]] = None,
+        response_type: Optional[Type[Response]] = None,
         middleware: Union[object, Iterable[object]] = None,
         router: Optional[routing.CompiledRouter] = None,
         independent_middleware: bool = True,
@@ -376,8 +376,8 @@ def __init__(
     ) -> None:
         super().__init__(
             media_type,
-            request_type,
-            response_type,
+            request_type or Request,
+            response_type or Response,
             middleware,
             router,
             independent_middleware,
@@ -386,7 +386,7 @@ def __init__(
         )
 
         self.ws_options = WebSocketOptions()
-        self._standard_response_type = response_type is Response
+        self._standard_response_type = response_type in (None, Response)
 
         self.add_error_handler(
             WebSocketDisconnected, self._ws_disconnected_error_handler
@@ -556,7 +556,7 @@ async def __call__(  # type: ignore[override] # noqa: C901
                     if data is None and resp._media is not None:
                         # NOTE(kgriffs): We use a special MISSING singleton since
                         #   None is ambiguous (the media handler might return None).
-                        if resp._media_rendered is MISSING:
+                        if resp._media_rendered is _UNSET:
                             opt = resp.options
                             if not resp.content_type:
                                 resp.content_type = opt.default_media_type