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