Skip to content

Repository navigation

Typing: use wsgiref.types to validate types and fix issues uncovered #2467

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged
lovelydinosaur merged 5 commits into from
Nov 29, 2022
Merged

Typing: use wsgiref.types to validate types and fix issues uncovered #2467

Changes from all commits
Commits
Show all changes
5 commits
Select commit Hold shift + click to select a range
f49a40a
Typing: use wsgiref.types to validate types and fix issues uncovered
mjpieters Nov 9, 2022
4d4f233
Merge remote-tracking branch 'upstream/master' into fix-wsgi
adriangb Nov 29, 2022
ead774f
Merge branch 'master' into fix-wsgi
adriangb Nov 29, 2022
b3526aa
remove typing_extensions
adriangb Nov 29, 2022
5f39804
Merge branch 'fix-wsgi' of https://github.com/adriangb/httpx into fix…
adriangb Nov 29, 2022
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
30 changes: 24 additions & 6 deletions httpx/_transports/wsgi.py
View file
Open in desktop
Original file line number Diff line number Diff line change
@@ -1,14 +1,31 @@
import io
import itertools
import sys
import types
import typing

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]]

def _skip_leading_empty_chunks(body: typing.Iterable[bytes]) -> typing.Iterable[bytes]:

# 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[
[typing.Dict[str, typing.Any], StartResponse], typing.Iterable[bytes]
]
Comment on lines +11 to +25

@michaeloliverx michaeloliverx Nov 29, 2022

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Hmm shouldn't we just import these directly from typeshed instead?
E.g.

if typing.TYPE_CHECKING:
    from _typeshed import OptExcInfo
    from _typeshed.wsgi import WSGIApplication

@adriangb adriangb Nov 29, 2022

Copy link
Copy Markdown
Contributor Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Probably. With an else: WSGIApplication = Any or something. Maybe make a PR?



def _skip_leading_empty_chunks(body: typing.Iterable[_T]) -> typing.Iterable[_T]:
body = iter(body)
for chunk in body:
if chunk:
Expand Down Expand Up @@ -54,7 +71,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.
Expand All @@ -65,7 +82,7 @@ class WSGITransport(BaseTransport):

def __init__(
self,
app: typing.Callable[..., typing.Any],
app: WSGIApplication,
raise_app_exceptions: bool = True,
script_name: str = "",
remote_addr: str = "127.0.0.1",
Expand Down Expand Up @@ -111,20 +128,21 @@ def handle_request(self, request: Request) -> Response:
def start_response(
status: str,
response_headers: typing.List[typing.Tuple[str, str]],
exc_info: typing.Any = None,
) -> None:
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)

stream = WSGIByteStream(result)

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])
Expand Down