Skip to content
Merged
Changes from all commits
Commits
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
23 changes: 6 additions & 17 deletions httpx/_transports/wsgi.py
Original file line number Diff line number Diff line change
@@ -1,28 +1,17 @@
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]]


# 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],
]
if typing.TYPE_CHECKING:
from _typeshed import OptExcInfo # pragma: no cover
from _typeshed.wsgi import WSGIApplication # pragma: no cover


WSGIApplication = typing.Callable[
[typing.Dict[str, typing.Any], StartResponse], typing.Iterable[bytes]
]
_T = typing.TypeVar("_T")


def _skip_leading_empty_chunks(body: typing.Iterable[_T]) -> typing.Iterable[_T]:
Expand Down Expand Up @@ -82,7 +71,7 @@ class WSGITransport(BaseTransport):

def __init__(
self,
app: WSGIApplication,
app: "WSGIApplication",
raise_app_exceptions: bool = True,
script_name: str = "",
remote_addr: str = "127.0.0.1",
Expand Down Expand Up @@ -128,7 +117,7 @@ def handle_request(self, request: Request) -> Response:
def start_response(
status: str,
response_headers: typing.List[typing.Tuple[str, str]],
exc_info: typing.Optional[_OptExcInfo] = None,
exc_info: typing.Optional["OptExcInfo"] = None,
) -> typing.Callable[[bytes], typing.Any]:
nonlocal seen_status, seen_response_headers, seen_exc_info
seen_status = status
Expand Down