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 contextlib import contextmanager
2from typing import Iterator, Optional, Union
4from .._models import (
5 URL,
6 Extensions,
7 HeaderTypes,
8 Origin,
9 Request,
10 Response,
11 enforce_bytes,
12 enforce_headers,
13 enforce_url,
14 include_request_headers,
15)
18class RequestInterface:
19 def request(
20 self,
21 method: Union[bytes, str],
22 url: Union[URL, bytes, str],
23 *,
24 headers: HeaderTypes = None,
25 content: Union[bytes, Iterator[bytes], None] = None,
26 extensions: Optional[Extensions] = None,
27 ) -> Response:
28 # Strict type checking on our parameters.
29 method = enforce_bytes(method, name="method")
30 url = enforce_url(url, name="url")
31 headers = enforce_headers(headers, name="headers")
33 # Include Host header, and optionally Content-Length or Transfer-Encoding.
34 headers = include_request_headers(headers, url=url, content=content)
36 request = Request(
37 method=method,
38 url=url,
39 headers=headers,
40 content=content,
41 extensions=extensions,
42 )
43 response = self.handle_request(request)
44 try:
45 response.read()
46 finally:
47 response.close()
48 return response
50 @contextmanager
51 def stream(
52 self,
53 method: Union[bytes, str],
54 url: Union[URL, bytes, str],
55 *,
56 headers: HeaderTypes = None,
57 content: Union[bytes, Iterator[bytes], None] = None,
58 extensions: Optional[Extensions] = None,
59 ) -> Iterator[Response]:
60 # Strict type checking on our parameters.
61 method = enforce_bytes(method, name="method")
62 url = enforce_url(url, name="url")
63 headers = enforce_headers(headers, name="headers")
65 # Include Host header, and optionally Content-Length or Transfer-Encoding.
66 headers = include_request_headers(headers, url=url, content=content)
68 request = Request(
69 method=method,
70 url=url,
71 headers=headers,
72 content=content,
73 extensions=extensions,
74 )
75 response = self.handle_request(request)
76 try:
77 yield response
78 finally:
79 response.close()
81 def handle_request(self, request: Request) -> Response:
82 raise NotImplementedError() # pragma: nocover
85class ConnectionInterface(RequestInterface):
86 def close(self) -> None:
87 raise NotImplementedError() # pragma: nocover
89 def info(self) -> str:
90 raise NotImplementedError() # pragma: nocover
92 def can_handle_request(self, origin: Origin) -> bool:
93 raise NotImplementedError() # pragma: nocover
95 def is_available(self) -> bool:
96 """
97 Return `True` if the connection is currently able to accept an
98 outgoing request.
100 An HTTP/1.1 connection will only be available if it is currently idle.
102 An HTTP/2 connection will be available so long as the stream ID space is
103 not yet exhausted, and the connection is not in an error state.
105 While the connection is being established we may not yet know if it is going
106 to result in an HTTP/1.1 or HTTP/2 connection. The connection should be
107 treated as being available, but might ultimately raise `NewConnectionRequired`
108 required exceptions if multiple requests are attempted over a connection
109 that ends up being established as HTTP/1.1.
110 """
111 raise NotImplementedError() # pragma: nocover
113 def has_expired(self) -> bool:
114 """
115 Return `True` if the connection is in a state where it should be closed.
117 This either means that the connection is idle and it has passed the
118 expiry time on its keep-alive, or that server has sent an EOF.
119 """
120 raise NotImplementedError() # pragma: nocover
122 def is_idle(self) -> bool:
123 """
124 Return `True` if the connection is currently idle.
125 """
126 raise NotImplementedError() # pragma: nocover
128 def is_closed(self) -> bool:
129 """
130 Return `True` if the connection has been closed.
132 Used when a response is closed to determine if the connection may be
133 returned to the connection pool or not.
134 """
135 raise NotImplementedError() # pragma: nocover