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