Spaces:
Sleeping
Sleeping
File size: 4,825 Bytes
7d134e4 |
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 131 132 133 134 135 136 137 138 139 140 141 142 143 144 145 146 147 148 149 150 |
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 # pragma: no cover
from _typeshed.wsgi import WSGIApplication # pragma: no cover
_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)
|