Coverage for /pythoncovmergedfiles/medio/medio/usr/local/lib/python3.11/site-packages/httpx/_transports/base.py: 74%

1from __future__ import annotations 

2 

3import typing 

4from types import TracebackType 

5 

6from .._models import Request, Response 

7 

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 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 

65class 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