| from __future__ import annotations |
|
|
| import io |
| import itertools |
| import sys |
| import typing |
|
|
| from .._models import Request, Response |
| from .._types import SyncByteStream |
| from .base import BaseTransport |
|
|
| if typing.TYPE_CHECKING: |
| from _typeshed import OptExcInfo |
| from _typeshed.wsgi import WSGIApplication |
|
|
| _T = typing.TypeVar("_T") |
|
|
|
|
| __all__ = ["WSGITransport"] |
|
|
|
|
| def _skip_leading_empty_chunks(body: typing.Iterable[_T]) -> typing.Iterable[_T]: |
| body = iter(body) |
| for chunk in body: |
| if chunk: |
| return itertools.chain([chunk], body) |
| return [] |
|
|
|
|
| class WSGIByteStream(SyncByteStream): |
| def __init__(self, result: typing.Iterable[bytes]) -> None: |
| self._close = getattr(result, "close", None) |
| self._result = _skip_leading_empty_chunks(result) |
|
|
| def __iter__(self) -> typing.Iterator[bytes]: |
| for part in self._result: |
| yield part |
|
|
| def close(self) -> None: |
| if self._close is not None: |
| self._close() |
|
|
|
|
| class WSGITransport(BaseTransport): |
| """ |
| A custom transport that handles sending requests directly to an WSGI app. |
| The simplest way to use this functionality is to use the `app` argument. |
| |
| ``` |
| client = httpx.Client(app=app) |
| ``` |
| |
| Alternatively, you can setup the transport instance explicitly. |
| This allows you to include any additional configuration arguments specific |
| to the WSGITransport class: |
| |
| ``` |
| transport = httpx.WSGITransport( |
| app=app, |
| script_name="/submount", |
| remote_addr="1.2.3.4" |
| ) |
| client = httpx.Client(transport=transport) |
| ``` |
| |
| Arguments: |
| |
| * `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. |
| * `script_name` - The root path on which the WSGI application should be mounted. |
| * `remote_addr` - A string indicating the client IP of incoming requests. |
| ``` |
| """ |
|
|
| def __init__( |
| self, |
| app: WSGIApplication, |
| raise_app_exceptions: bool = True, |
| script_name: str = "", |
| remote_addr: str = "127.0.0.1", |
| wsgi_errors: typing.TextIO | None = None, |
| ) -> None: |
| self.app = app |
| self.raise_app_exceptions = raise_app_exceptions |
| self.script_name = script_name |
| self.remote_addr = remote_addr |
| self.wsgi_errors = wsgi_errors |
|
|
| def handle_request(self, request: Request) -> Response: |
| request.read() |
| wsgi_input = io.BytesIO(request.content) |
|
|
| port = request.url.port or {"http": 80, "https": 443}[request.url.scheme] |
| environ = { |
| "wsgi.version": (1, 0), |
| "wsgi.url_scheme": request.url.scheme, |
| "wsgi.input": wsgi_input, |
| "wsgi.errors": self.wsgi_errors or sys.stderr, |
| "wsgi.multithread": True, |
| "wsgi.multiprocess": False, |
| "wsgi.run_once": False, |
| "REQUEST_METHOD": request.method, |
| "SCRIPT_NAME": self.script_name, |
| "PATH_INFO": request.url.path, |
| "QUERY_STRING": request.url.query.decode("ascii"), |
| "SERVER_NAME": request.url.host, |
| "SERVER_PORT": str(port), |
| "SERVER_PROTOCOL": "HTTP/1.1", |
| "REMOTE_ADDR": self.remote_addr, |
| } |
| for header_key, header_value in request.headers.raw: |
| key = header_key.decode("ascii").upper().replace("-", "_") |
| if key not in ("CONTENT_TYPE", "CONTENT_LENGTH"): |
| key = "HTTP_" + key |
| environ[key] = header_value.decode("ascii") |
|
|
| seen_status = None |
| seen_response_headers = None |
| seen_exc_info = None |
|
|
| def start_response( |
| status: str, |
| response_headers: list[tuple[str, str]], |
| exc_info: OptExcInfo | None = 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 seen_exc_info[0] and self.raise_app_exceptions: |
| raise seen_exc_info[1] |
|
|
| status_code = int(seen_status.split()[0]) |
| headers = [ |
| (key.encode("ascii"), value.encode("ascii")) |
| for key, value in seen_response_headers |
| ] |
|
|
| return Response(status_code, headers=headers, stream=stream) |
|
|
