codekingpro/portable-devtools

1from __future__ import annotations2 3import typing4from types import TracebackType5 6from .._models import Request, Response7 8T = typing.TypeVar("T", bound="BaseTransport")9A = typing.TypeVar("A", bound="AsyncBaseTransport")10 11__all__ = ["AsyncBaseTransport", "BaseTransport"]12 13 14class BaseTransport:15    def __enter__(self: T) -> T:16        return self17 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 API32        niceties.33 34        In order to properly release any network resources, the response35        stream should *either* be consumed immediately, with a call to36        `response.stream.read()`, or else the `handle_request` call should37        be followed with a try/finally block to ensuring the stream is38        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 cover60 61    def close(self) -> None:62        pass63 64 65class AsyncBaseTransport:66    async def __aenter__(self: A) -> A:67        return self68 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 cover84 85    async def aclose(self) -> None:86        pass87