Coverage for /pythoncovmergedfiles/medio/medio/usr/local/lib/python3.8/site-packages/httpcore/_models.py: 28%
148 statements
« prev ^ index » next coverage.py v7.2.2, created at 2023-03-26 06:12 +0000
1from typing import (
2 Any,
3 AsyncIterable,
4 AsyncIterator,
5 Iterable,
6 Iterator,
7 List,
8 Mapping,
9 Optional,
10 Sequence,
11 Tuple,
12 Union,
13)
14from urllib.parse import urlparse
16# Functions for typechecking...
19HeadersAsSequence = Sequence[Tuple[Union[bytes, str], Union[bytes, str]]]
20HeadersAsMapping = Mapping[Union[bytes, str], Union[bytes, str]]
21HeaderTypes = Union[HeadersAsSequence, HeadersAsMapping, None]
23Extensions = Mapping[str, Any]
26def enforce_bytes(value: Union[bytes, str], *, name: str) -> bytes:
27 """
28 Any arguments that are ultimately represented as bytes can be specified
29 either as bytes or as strings.
31 However we enforce that any string arguments must only contain characters in
32 the plain ASCII range. chr(0)...chr(127). If you need to use characters
33 outside that range then be precise, and use a byte-wise argument.
34 """
35 if isinstance(value, str):
36 try:
37 return value.encode("ascii")
38 except UnicodeEncodeError:
39 raise TypeError(f"{name} strings may not include unicode characters.")
40 elif isinstance(value, bytes):
41 return value
43 seen_type = type(value).__name__
44 raise TypeError(f"{name} must be bytes or str, but got {seen_type}.")
47def enforce_url(value: Union["URL", bytes, str], *, name: str) -> "URL":
48 """
49 Type check for URL parameters.
50 """
51 if isinstance(value, (bytes, str)):
52 return URL(value)
53 elif isinstance(value, URL):
54 return value
56 seen_type = type(value).__name__
57 raise TypeError(f"{name} must be a URL, bytes, or str, but got {seen_type}.")
60def enforce_headers(
61 value: Union[HeadersAsMapping, HeadersAsSequence, None] = None, *, name: str
62) -> List[Tuple[bytes, bytes]]:
63 """
64 Convienence function that ensure all items in request or response headers
65 are either bytes or strings in the plain ASCII range.
66 """
67 if value is None:
68 return []
69 elif isinstance(value, Mapping):
70 return [
71 (
72 enforce_bytes(k, name="header name"),
73 enforce_bytes(v, name="header value"),
74 )
75 for k, v in value.items()
76 ]
77 elif isinstance(value, Sequence):
78 return [
79 (
80 enforce_bytes(k, name="header name"),
81 enforce_bytes(v, name="header value"),
82 )
83 for k, v in value
84 ]
86 seen_type = type(value).__name__
87 raise TypeError(
88 f"{name} must be a mapping or sequence of two-tuples, but got {seen_type}."
89 )
92def enforce_stream(
93 value: Union[bytes, Iterable[bytes], AsyncIterable[bytes], None], *, name: str
94) -> Union[Iterable[bytes], AsyncIterable[bytes]]:
95 if value is None:
96 return ByteStream(b"")
97 elif isinstance(value, bytes):
98 return ByteStream(value)
99 return value
102# * https://tools.ietf.org/html/rfc3986#section-3.2.3
103# * https://url.spec.whatwg.org/#url-miscellaneous
104# * https://url.spec.whatwg.org/#scheme-state
105DEFAULT_PORTS = {
106 b"ftp": 21,
107 b"http": 80,
108 b"https": 443,
109 b"ws": 80,
110 b"wss": 443,
111}
114def include_request_headers(
115 headers: List[Tuple[bytes, bytes]],
116 *,
117 url: "URL",
118 content: Union[None, bytes, Iterable[bytes], AsyncIterable[bytes]],
119) -> List[Tuple[bytes, bytes]]:
120 headers_set = set(k.lower() for k, v in headers)
122 if b"host" not in headers_set:
123 default_port = DEFAULT_PORTS.get(url.scheme)
124 if url.port is None or url.port == default_port:
125 header_value = url.host
126 else:
127 header_value = b"%b:%d" % (url.host, url.port)
128 headers = [(b"Host", header_value)] + headers
130 if (
131 content is not None
132 and b"content-length" not in headers_set
133 and b"transfer-encoding" not in headers_set
134 ):
135 if isinstance(content, bytes):
136 content_length = str(len(content)).encode("ascii")
137 headers += [(b"Content-Length", content_length)]
138 else:
139 headers += [(b"Transfer-Encoding", b"chunked")] # pragma: nocover
141 return headers
144# Interfaces for byte streams...
147class ByteStream:
148 """
149 A container for non-streaming content, and that supports both sync and async
150 stream iteration.
151 """
153 def __init__(self, content: bytes) -> None:
154 self._content = content
156 def __iter__(self) -> Iterator[bytes]:
157 yield self._content
159 async def __aiter__(self) -> AsyncIterator[bytes]:
160 yield self._content
162 def __repr__(self) -> str:
163 return f"<{self.__class__.__name__} [{len(self._content)} bytes]>"
166class Origin:
167 def __init__(self, scheme: bytes, host: bytes, port: int) -> None:
168 self.scheme = scheme
169 self.host = host
170 self.port = port
172 def __eq__(self, other: Any) -> bool:
173 return (
174 isinstance(other, Origin)
175 and self.scheme == other.scheme
176 and self.host == other.host
177 and self.port == other.port
178 )
180 def __str__(self) -> str:
181 scheme = self.scheme.decode("ascii")
182 host = self.host.decode("ascii")
183 port = str(self.port)
184 return f"{scheme}://{host}:{port}"
187class URL:
188 """
189 Represents the URL against which an HTTP request may be made.
191 The URL may either be specified as a plain string, for convienence:
193 ```python
194 url = httpcore.URL("https://www.example.com/")
195 ```
197 Or be constructed with explicitily pre-parsed components:
199 ```python
200 url = httpcore.URL(scheme=b'https', host=b'www.example.com', port=None, target=b'/')
201 ```
203 Using this second more explicit style allows integrations that are using
204 `httpcore` to pass through URLs that have already been parsed in order to use
205 libraries such as `rfc-3986` rather than relying on the stdlib. It also ensures
206 that URL parsing is treated identically at both the networking level and at any
207 higher layers of abstraction.
209 The four components are important here, as they allow the URL to be precisely
210 specified in a pre-parsed format. They also allow certain types of request to
211 be created that could not otherwise be expressed.
213 For example, an HTTP request to `http://www.example.com/` forwarded via a proxy
214 at `http://localhost:8080`...
216 ```python
217 # Constructs an HTTP request with a complete URL as the target:
218 # GET https://www.example.com/ HTTP/1.1
219 url = httpcore.URL(
220 scheme=b'http',
221 host=b'localhost',
222 port=8080,
223 target=b'https://www.example.com/'
224 )
225 request = httpcore.Request(
226 method="GET",
227 url=url
228 )
229 ```
231 Another example is constructing an `OPTIONS *` request...
233 ```python
234 # Constructs an 'OPTIONS *' HTTP request:
235 # OPTIONS * HTTP/1.1
236 url = httpcore.URL(scheme=b'https', host=b'www.example.com', target=b'*')
237 request = httpcore.Request(method="OPTIONS", url=url)
238 ```
240 This kind of request is not possible to formulate with a URL string,
241 because the `/` delimiter is always used to demark the target from the
242 host/port portion of the URL.
244 For convenience, string-like arguments may be specified either as strings or
245 as bytes. However, once a request is being issue over-the-wire, the URL
246 components are always ultimately required to be a bytewise representation.
248 In order to avoid any ambiguity over character encodings, when strings are used
249 as arguments, they must be strictly limited to the ASCII range `chr(0)`-`chr(127)`.
250 If you require a bytewise representation that is outside this range you must
251 handle the character encoding directly, and pass a bytes instance.
252 """
254 def __init__(
255 self,
256 url: Union[bytes, str] = "",
257 *,
258 scheme: Union[bytes, str] = b"",
259 host: Union[bytes, str] = b"",
260 port: Optional[int] = None,
261 target: Union[bytes, str] = b"",
262 ) -> None:
263 """
264 Parameters:
265 url: The complete URL as a string or bytes.
266 scheme: The URL scheme as a string or bytes.
267 Typically either `"http"` or `"https"`.
268 host: The URL host as a string or bytes. Such as `"www.example.com"`.
269 port: The port to connect to. Either an integer or `None`.
270 target: The target of the HTTP request. Such as `"/items?search=red"`.
271 """
272 if url:
273 parsed = urlparse(enforce_bytes(url, name="url"))
274 self.scheme = parsed.scheme
275 self.host = parsed.hostname or b""
276 self.port = parsed.port
277 self.target = (parsed.path or b"/") + (
278 b"?" + parsed.query if parsed.query else b""
279 )
280 else:
281 self.scheme = enforce_bytes(scheme, name="scheme")
282 self.host = enforce_bytes(host, name="host")
283 self.port = port
284 self.target = enforce_bytes(target, name="target")
286 @property
287 def origin(self) -> Origin:
288 default_port = {
289 b"http": 80,
290 b"https": 443,
291 b"ws": 80,
292 b"wss": 443,
293 b"socks5": 1080,
294 }[self.scheme]
295 return Origin(
296 scheme=self.scheme, host=self.host, port=self.port or default_port
297 )
299 def __eq__(self, other: Any) -> bool:
300 return (
301 isinstance(other, URL)
302 and other.scheme == self.scheme
303 and other.host == self.host
304 and other.port == self.port
305 and other.target == self.target
306 )
308 def __bytes__(self) -> bytes:
309 if self.port is None:
310 return b"%b://%b%b" % (self.scheme, self.host, self.target)
311 return b"%b://%b:%d%b" % (self.scheme, self.host, self.port, self.target)
313 def __repr__(self) -> str:
314 return (
315 f"{self.__class__.__name__}(scheme={self.scheme!r}, "
316 f"host={self.host!r}, port={self.port!r}, target={self.target!r})"
317 )
320class Request:
321 """
322 An HTTP request.
323 """
325 def __init__(
326 self,
327 method: Union[bytes, str],
328 url: Union[URL, bytes, str],
329 *,
330 headers: HeaderTypes = None,
331 content: Union[bytes, Iterable[bytes], AsyncIterable[bytes], None] = None,
332 extensions: Optional[Extensions] = None,
333 ) -> None:
334 """
335 Parameters:
336 method: The HTTP request method, either as a string or bytes.
337 For example: `GET`.
338 url: The request URL, either as a `URL` instance, or as a string or bytes.
339 For example: `"https://www.example.com".`
340 headers: The HTTP request headers.
341 content: The content of the response body.
342 extensions: A dictionary of optional extra information included on
343 the request. Possible keys include `"timeout"`, and `"trace"`.
344 """
345 self.method: bytes = enforce_bytes(method, name="method")
346 self.url: URL = enforce_url(url, name="url")
347 self.headers: List[Tuple[bytes, bytes]] = enforce_headers(
348 headers, name="headers"
349 )
350 self.stream: Union[Iterable[bytes], AsyncIterable[bytes]] = enforce_stream(
351 content, name="content"
352 )
353 self.extensions = {} if extensions is None else extensions
355 def __repr__(self) -> str:
356 return f"<{self.__class__.__name__} [{self.method!r}]>"
359class Response:
360 """
361 An HTTP response.
362 """
364 def __init__(
365 self,
366 status: int,
367 *,
368 headers: HeaderTypes = None,
369 content: Union[bytes, Iterable[bytes], AsyncIterable[bytes], None] = None,
370 extensions: Optional[Extensions] = None,
371 ) -> None:
372 """
373 Parameters:
374 status: The HTTP status code of the response. For example `200`.
375 headers: The HTTP response headers.
376 content: The content of the response body.
377 extensions: A dictionary of optional extra information included on
378 the responseself.Possible keys include `"http_version"`,
379 `"reason_phrase"`, and `"network_stream"`.
380 """
381 self.status: int = status
382 self.headers: List[Tuple[bytes, bytes]] = enforce_headers(
383 headers, name="headers"
384 )
385 self.stream: Union[Iterable[bytes], AsyncIterable[bytes]] = enforce_stream(
386 content, name="content"
387 )
388 self.extensions = {} if extensions is None else extensions
390 self._stream_consumed = False
392 @property
393 def content(self) -> bytes:
394 if not hasattr(self, "_content"):
395 if isinstance(self.stream, Iterable):
396 raise RuntimeError(
397 "Attempted to access 'response.content' on a streaming response. "
398 "Call 'response.read()' first."
399 )
400 else:
401 raise RuntimeError(
402 "Attempted to access 'response.content' on a streaming response. "
403 "Call 'await response.aread()' first."
404 )
405 return self._content
407 def __repr__(self) -> str:
408 return f"<{self.__class__.__name__} [{self.status}]>"
410 # Sync interface...
412 def read(self) -> bytes:
413 if not isinstance(self.stream, Iterable): # pragma: nocover
414 raise RuntimeError(
415 "Attempted to read an asynchronous response using 'response.read()'. "
416 "You should use 'await response.aread()' instead."
417 )
418 if not hasattr(self, "_content"):
419 self._content = b"".join([part for part in self.iter_stream()])
420 return self._content
422 def iter_stream(self) -> Iterator[bytes]:
423 if not isinstance(self.stream, Iterable): # pragma: nocover
424 raise RuntimeError(
425 "Attempted to stream an asynchronous response using 'for ... in "
426 "response.iter_stream()'. "
427 "You should use 'async for ... in response.aiter_stream()' instead."
428 )
429 if self._stream_consumed:
430 raise RuntimeError(
431 "Attempted to call 'for ... in response.iter_stream()' more than once."
432 )
433 self._stream_consumed = True
434 for chunk in self.stream:
435 yield chunk
437 def close(self) -> None:
438 if not isinstance(self.stream, Iterable): # pragma: nocover
439 raise RuntimeError(
440 "Attempted to close an asynchronous response using 'response.close()'. "
441 "You should use 'await response.aclose()' instead."
442 )
443 if hasattr(self.stream, "close"):
444 self.stream.close()
446 # Async interface...
448 async def aread(self) -> bytes:
449 if not isinstance(self.stream, AsyncIterable): # pragma: nocover
450 raise RuntimeError(
451 "Attempted to read an synchronous response using "
452 "'await response.aread()'. "
453 "You should use 'response.read()' instead."
454 )
455 if not hasattr(self, "_content"):
456 self._content = b"".join([part async for part in self.aiter_stream()])
457 return self._content
459 async def aiter_stream(self) -> AsyncIterator[bytes]:
460 if not isinstance(self.stream, AsyncIterable): # pragma: nocover
461 raise RuntimeError(
462 "Attempted to stream an synchronous response using 'async for ... in "
463 "response.aiter_stream()'. "
464 "You should use 'for ... in response.iter_stream()' instead."
465 )
466 if self._stream_consumed:
467 raise RuntimeError(
468 "Attempted to call 'async for ... in response.aiter_stream()' "
469 "more than once."
470 )
471 self._stream_consumed = True
472 async for chunk in self.stream:
473 yield chunk
475 async def aclose(self) -> None:
476 if not isinstance(self.stream, AsyncIterable): # pragma: nocover
477 raise RuntimeError(
478 "Attempted to close a synchronous response using "
479 "'await response.aclose()'. "
480 "You should use 'response.close()' instead."
481 )
482 if hasattr(self.stream, "aclose"):
483 await self.stream.aclose()