From f49a40aa615e39f4c56d3d78ec3d3ad67d2fba58 Mon Sep 17 00:00:00 2001 From: Martijn Pieters Date: Wed, 9 Nov 2022 22:48:04 +0000 Subject: [PATCH 1/2] Typing: use wsgiref.types to validate types and fix issues uncovered - start_response() must return a write(bytes) function, even though this is now deprecated. It's fine to be a no-op here. - sys.exc_info() can return (None, None, None), so make sure to handle that case. --- httpx/_transports/wsgi.py | 39 +++++++++++++++++++++++++++++++++------ 1 file changed, 33 insertions(+), 6 deletions(-) diff --git a/httpx/_transports/wsgi.py b/httpx/_transports/wsgi.py index 3dedf49f96..f2d53b11c1 100644 --- a/httpx/_transports/wsgi.py +++ b/httpx/_transports/wsgi.py @@ -1,14 +1,38 @@ import io import itertools import sys +import types import typing +import typing_extensions as te + from .._models import Request, Response from .._types import SyncByteStream from .base import BaseTransport +_T = typing.TypeVar("_T") +_ExcInfo = typing.Tuple[typing.Type[BaseException], BaseException, types.TracebackType] +_OptExcInfo = typing.Union[_ExcInfo, typing.Tuple[None, None, None]] + +# backported wsgiref.types definitions from Python 3.11 + + +class StartResponse(te.Protocol): + def __call__( + self, + __status: str, + __headers: typing.List[typing.Tuple[str, str]], + __exc_info: typing.Optional[_OptExcInfo] = ..., + ) -> typing.Callable[[bytes], object]: # pragma: no cover + ... + + +WSGIApplication = typing.Callable[ + [typing.Dict[str, typing.Any], StartResponse], typing.Iterable[bytes] +] + -def _skip_leading_empty_chunks(body: typing.Iterable) -> typing.Iterable: +def _skip_leading_empty_chunks(body: typing.Iterable[_T]) -> typing.Iterable[_T]: body = iter(body) for chunk in body: if chunk: @@ -54,7 +78,7 @@ class WSGITransport(BaseTransport): Arguments: - * `app` - The ASGI application. + * `app` - The WSGI application. * `raise_app_exceptions` - Boolean indicating if exceptions in the application should be raised. Default to `True`. Can be set to `False` for use cases such as testing the content of a client 500 response. @@ -65,7 +89,7 @@ class WSGITransport(BaseTransport): def __init__( self, - app: typing.Callable, + app: WSGIApplication, raise_app_exceptions: bool = True, script_name: str = "", remote_addr: str = "127.0.0.1", @@ -109,12 +133,15 @@ def handle_request(self, request: Request) -> Response: seen_exc_info = None def start_response( - status: str, response_headers: list, exc_info: typing.Any = None - ) -> None: + status: str, + response_headers: typing.List[typing.Tuple[str, str]], + exc_info: typing.Optional[_OptExcInfo] = None, + ) -> typing.Callable[[bytes], typing.Any]: nonlocal seen_status, seen_response_headers, seen_exc_info seen_status = status seen_response_headers = response_headers seen_exc_info = exc_info + return lambda _: None result = self.app(environ, start_response) @@ -122,7 +149,7 @@ def start_response( assert seen_status is not None assert seen_response_headers is not None - if seen_exc_info and self.raise_app_exceptions: + if seen_exc_info and seen_exc_info[0] and self.raise_app_exceptions: raise seen_exc_info[1] status_code = int(seen_status.split()[0]) From b3526aa7a961dcf3b5f031bec8e881384bb1f37b Mon Sep 17 00:00:00 2001 From: Adrian Garcia Badaracco <1755071+adriangb@users.noreply.github.com> Date: Tue, 29 Nov 2022 11:51:18 -0600 Subject: [PATCH 2/2] remove typing_extensions --- httpx/_transports/wsgi.py | 17 +++++------------ 1 file changed, 5 insertions(+), 12 deletions(-) diff --git a/httpx/_transports/wsgi.py b/httpx/_transports/wsgi.py index f2d53b11c1..f27a77aea6 100644 --- a/httpx/_transports/wsgi.py +++ b/httpx/_transports/wsgi.py @@ -4,8 +4,6 @@ import types import typing -import typing_extensions as te - from .._models import Request, Response from .._types import SyncByteStream from .base import BaseTransport @@ -14,17 +12,12 @@ _ExcInfo = typing.Tuple[typing.Type[BaseException], BaseException, types.TracebackType] _OptExcInfo = typing.Union[_ExcInfo, typing.Tuple[None, None, None]] -# backported wsgiref.types definitions from Python 3.11 - -class StartResponse(te.Protocol): - def __call__( - self, - __status: str, - __headers: typing.List[typing.Tuple[str, str]], - __exc_info: typing.Optional[_OptExcInfo] = ..., - ) -> typing.Callable[[bytes], object]: # pragma: no cover - ... +# backported wsgiref.types definitions from Python 3.11 +StartResponse = typing.Callable[ + [str, typing.List[typing.Tuple[str, str]], typing.Optional[_OptExcInfo]], + typing.Callable[[bytes], object], +] WSGIApplication = typing.Callable[