Shortcuts on this page
r m x toggle line displays
j k next/prev highlighted chunk
0 (zero) top of page
1 (one) first highlighted chunk
1from __future__ import annotations
3import typing
4from types import TracebackType
6from .._models import Request, Response
8T = typing.TypeVar("T", bound="BaseTransport")
9A = typing.TypeVar("A", bound="AsyncBaseTransport")
12class BaseTransport:
13 def __enter__(self: T) -> T:
14 return self
16 def __exit__(
17 self,
18 exc_type: type[BaseException] | None = None,
19 exc_value: BaseException | None = None,
20 traceback: TracebackType | None = None,
21 ) -> None:
22 self.close()
24 def handle_request(self, request: Request) -> Response:
25 """
26 Send a single HTTP request and return a response.
28 Developers shouldn't typically ever need to call into this API directly,
29 since the Client class provides all the higher level user-facing API
30 niceties.
32 In order to properly release any network resources, the response
33 stream should *either* be consumed immediately, with a call to
34 `response.stream.read()`, or else the `handle_request` call should
35 be followed with a try/finally block to ensuring the stream is
36 always closed.
38 Example usage:
40 with httpx.HTTPTransport() as transport:
41 req = httpx.Request(
42 method=b"GET",
43 url=(b"https", b"www.example.com", 443, b"/"),
44 headers=[(b"Host", b"www.example.com")],
45 )
46 resp = transport.handle_request(req)
47 body = resp.stream.read()
48 print(resp.status_code, resp.headers, body)
51 Takes a `Request` instance as the only argument.
53 Returns a `Response` instance.
54 """
55 raise NotImplementedError(
56 "The 'handle_request' method must be implemented."
57 ) # pragma: no cover
59 def close(self) -> None:
60 pass
63class AsyncBaseTransport:
64 async def __aenter__(self: A) -> A:
65 return self
67 async def __aexit__(
68 self,
69 exc_type: type[BaseException] | None = None,
70 exc_value: BaseException | None = None,
71 traceback: TracebackType | None = None,
72 ) -> None:
73 await self.aclose()
75 async def handle_async_request(
76 self,
77 request: Request,
78 ) -> Response:
79 raise NotImplementedError(
80 "The 'handle_async_request' method must be implemented."
81 ) # pragma: no cover
83 async def aclose(self) -> None:
84 pass