From 75cc76ef17a604a08ee080898d5a082e3c3ec0de Mon Sep 17 00:00:00 2001 From: Kim Christie Date: Wed, 4 Nov 2026 21:41:52 +0100 Subject: [PATCH] First pass async support --- src/hip/_body.py | 16 +++--- src/hip/_client.py | 46 +++++++++--------- src/hip/_content.py | 28 +++++------ src/hip/_forms.py | 10 ++-- src/hip/_parsers.py | 62 ++++++++++++------------ src/hip/_server.py | 52 ++++++++++---------- src/hip/_streams.py | 24 ++++----- src/hip/backends/asyncio.py | 83 ++++++++++++++++++++++++++++++++ src/hip/transports/base.py | 16 +++--- src/hip/transports/connection.py | 62 +++++++++++------------- src/hip/transports/pool.py | 28 +++++------ 11 files changed, 253 insertions(+), 174 deletions(-) create mode 100644 src/hip/backends/asyncio.py diff --git a/src/hip/_body.py b/src/hip/_body.py index 5b1350d..8e09c6d 100644 --- a/src/hip/_body.py +++ b/src/hip/_body.py @@ -14,7 +14,7 @@ def __init__(self, parser: HTTPParser, is_response: bool): self._buffer = io.BytesIO() self._finished = False - def read(self, size=-1) -> bytes: + async def read(self, size=-1) -> bytes: if self._finished: return b"" @@ -31,7 +31,7 @@ def read(self, size=-1) -> bytes: # Read each chunk in turn. while (size < 0) or (length < size): - section = self._parser.recv_body() + section = await self._parser.recv_body() sections.append(section) length += len(section) if section == b"": @@ -47,10 +47,10 @@ def read(self, size=-1) -> bytes: return output - def close(self) -> None: + async def close(self) -> None: self._buffer.close() if self._is_response: - self._parser.complete() + await self._parser.complete() class GZipStream(Stream): @@ -61,9 +61,9 @@ def __init__(self, stream: Stream): self._finished = False self._buffer = bytearray() - def read(self, size=-1) -> bytes: + async def read(self, size=-1) -> bytes: while (size < 0 or len(self._buffer) < size) and not self._finished: - compressed = self._stream.read(8192) + compressed = await self._stream.read(8192) try: if compressed: @@ -88,5 +88,5 @@ def read(self, size=-1) -> bytes: return output - def close(self) -> None: - self._stream.close() + async def close(self) -> None: + await self._stream.close() diff --git a/src/hip/_client.py b/src/hip/_client.py index cbd355e..b8163ff 100644 --- a/src/hip/_client.py +++ b/src/hip/_client.py @@ -49,7 +49,7 @@ def build_request( content=content, ) - def request( + async def request( self, method: Method | str, url: URL | str, @@ -57,67 +57,67 @@ def request( content: Content | bytes | None = None, ) -> Response: request = self.build_request(method, url, headers=headers, content=content) - with self.via.send(request) as (status_code, resp_headers, stream): + async with await self.via.send(request) as (status_code, resp_headers, stream): ct = resp_headers.get("Content-Type", "application/octet-stream") cls = select_media(self.media, ct) - content = cls.parse(stream) + content = await cls.parse(stream) return Response( status_code=status_code, headers=resp_headers, content=content, ) - def get( + async def get( self, url: URL | str, headers: Headers | dict[str, str] | None = None, ): - return self.request("GET", url, headers=headers) + return await self.request("GET", url, headers=headers) - def post( + async def post( self, url: URL | str, headers: Headers | dict[str, str] | None = None, content: Content | bytes | None = None, ): - return self.request("POST", url, headers=headers, content=content) + return await self.request("POST", url, headers=headers, content=content) - def put( + async def put( self, url: URL | str, headers: Headers | dict[str, str] | None = None, content: Content | bytes | None = None, ): - return self.request("PUT", url, headers=headers, content=content) + return await self.request("PUT", url, headers=headers, content=content) - def patch( + async def patch( self, url: URL | str, headers: Headers | dict[str, str] | None = None, content: Content | bytes | None = None, ): - return self.request("PATCH", url, headers=headers, content=content) + return await self.request("PATCH", url, headers=headers, content=content) - def delete( + async def delete( self, url: URL | str, headers: Headers | dict[str, str] | None = None, ): - return self.request("DELETE", url, headers=headers) + return await self.request("DELETE", url, headers=headers) - def close(self): - self.via.close() + async def close(self): + await self.via.close() - def __enter__(self): + async def __aenter__(self): return self - def __exit__( + async def __aexit__( self, exc_type: type[BaseException] | None = None, exc_value: BaseException | None = None, traceback: types.TracebackType | None = None, ): - self.close() + await self.close() def __repr__(self): return f"" @@ -150,9 +150,9 @@ def build_redirect_request( return Request(method, url, headers, content) - def send(self, request: Request) -> SendContext: + async def send(self, request: Request) -> SendContext: while True: - ctx = self._transport.send(request) + ctx = await self._transport.send(request) # Determine if we have a redirect or not. redirect = self.build_redirect_request(request, ctx) @@ -166,10 +166,10 @@ def send(self, request: Request) -> SendContext: # request/response cycle. with ctx: _, _, stream = ctx - stream.read() + await stream.read() # Make the next request request = redirect - def close(self): - self._transport.close() + async def close(self): + await self._transport.close() diff --git a/src/hip/_content.py b/src/hip/_content.py index 852b7d0..d2ee4bd 100644 --- a/src/hip/_content.py +++ b/src/hip/_content.py @@ -17,7 +17,7 @@ class Content: - def open(self) -> Stream: + async def open(self) -> Stream: raise NotImplementedError() def headers(self) -> Headers: @@ -44,7 +44,7 @@ def __init__(self, data: typing.Any, source: bytes | None = None) -> None: else source ) - def open(self) -> Stream: + async def open(self) -> Stream: return ByteStream(self._content) def headers(self) -> Headers: @@ -58,8 +58,8 @@ def headers(self) -> Headers: ) @classmethod - def parse(self, stream: Stream) -> "Content": - source = stream.read() + async def parse(self, stream: Stream) -> "Content": + source = await stream.read() data = json.loads(source) return JSON(data, source) @@ -99,7 +99,7 @@ def __init__(self, text: str, source: bytes | None = None) -> None: self._text = text self._content = self._text.encode("utf-8") if source is None else source - def open(self) -> Stream: + async def open(self) -> Stream: return ByteStream(self._content) def headers(self) -> Headers: @@ -113,8 +113,8 @@ def headers(self) -> Headers: ) @classmethod - def parse(self, stream: Stream) -> Content: - source = stream.read() + async def parse(self, stream: Stream) -> Content: + source = await stream.read() text = source.decode("utf-8") return Text(text, source) @@ -133,7 +133,7 @@ def __init__(self, text: str, source: bytes | None = None) -> None: self._text = text self._content = self._text.encode("utf-8") if source is None else source - def open(self) -> Stream: + async def open(self) -> Stream: return ByteStream(self._content) def headers(self) -> Headers: @@ -147,8 +147,8 @@ def headers(self) -> Headers: ) @classmethod - def parse(self, stream: Stream) -> Content: - source = stream.read() + async def parse(self, stream: Stream) -> Content: + source = await stream.read() text = source.decode("utf-8") return HTML(text, source) @@ -166,7 +166,7 @@ class Binary(Content): def __init__(self, content: bytes | None) -> None: self._content = b"" if (content is None) else content - def open(self) -> Stream: + async def open(self) -> Stream: return ByteStream(self._content) def headers(self) -> Headers: @@ -174,8 +174,8 @@ def headers(self) -> Headers: return Headers({"Content-Length": f"{content_length}"}) @classmethod - def parse(self, stream: Stream) -> Content: - source = stream.read() + async def parse(self, stream: Stream) -> Content: + source = await stream.read() return Binary(source) def __bytes__(self) -> bytes: @@ -189,7 +189,7 @@ class Empty(Content): def __init__(self) -> None: pass - def open(self) -> Stream: + async def open(self) -> Stream: return ByteStream(b"") def headers(self) -> Headers: diff --git a/src/hip/_forms.py b/src/hip/_forms.py index c444fe9..248ecbd 100644 --- a/src/hip/_forms.py +++ b/src/hip/_forms.py @@ -62,7 +62,7 @@ def __init__( # Content API - def open(self) -> Stream: + async def open(self) -> Stream: return ByteStream(self._content) def headers(self) -> Headers: @@ -168,7 +168,7 @@ def __init__(self, path: str): def name(self) -> str: return os.path.basename(self._path) - def open(self) -> Stream: + async def open(self) -> Stream: fileobj = open(self._path, "rb") # noqa: SIM115 return FileStream(fileobj) @@ -254,8 +254,8 @@ def get_list(self, key: str) -> list[File]: return list(self._dict.get(key, [])) # Content interface - def open(self) -> Stream: - return MultiPart(files=self, boundary=self._boundary).open() + async def open(self) -> Stream: + return await MultiPart(files=self, boundary=self._boundary).open() def headers(self) -> Headers: content_type = f"multipart/form-data; boundary={self._boundary}" @@ -321,7 +321,7 @@ def form(self) -> Form: def files(self) -> Files: return self._files - def open(self) -> Stream: + async def open(self) -> Stream: form = [(key, value) for key, value in self._form.items()] files = [(key, file._path) for key, file in self._files.items()] return MultiPartStream(form, files, boundary=self._boundary) diff --git a/src/hip/_parsers.py b/src/hip/_parsers.py index 64391c0..1730691 100644 --- a/src/hip/_parsers.py +++ b/src/hip/_parsers.py @@ -93,7 +93,7 @@ def __init__(self, stream: Stream, mode: str) -> None: # Special states... self.processing_1xx = False - def send_method_line(self, method: bytes, target: bytes, protocol: bytes) -> None: + async def send_method_line(self, method: bytes, target: bytes, protocol: bytes) -> None: """ Send the initial request line: @@ -112,12 +112,12 @@ def send_method_line(self, method: bytes, target: bytes, protocol: bytes) -> Non if protocol != b"HTTP/1.1": raise ProtocolError("Sent unsupported protocol version") data = b" ".join([method, target, protocol]) + b"\r\n" - self.stream.write(data) + await self.stream.write(data) self.send_state = State.SEND_HEADERS self.recv_state = State.RECV_STATUS_LINE - def send_status_line( + async def send_status_line( self, protocol: bytes, status_code: int, reason: bytes ) -> None: """ @@ -136,11 +136,11 @@ def send_status_line( raise ProtocolError("Sent unsupported protocol version") status_code_bytes = str(status_code).encode("ascii") data = b" ".join([protocol, status_code_bytes, reason]) + b"\r\n" - self.stream.write(data) + await self.stream.write(data) self.send_state = State.SEND_HEADERS - def send_headers(self, headers: list[tuple[bytes, bytes]]) -> None: + async def send_headers(self, headers: list[tuple[bytes, bytes]]) -> None: """ Send the request headers: @@ -173,11 +173,11 @@ def send_headers(self, headers: list[tuple[bytes, bytes]]) -> None: # Send request headers lines = [name + b": " + value + b"\r\n" for name, value in headers] data = b"".join(lines) + b"\r\n" - self.stream.write(data) + await self.stream.write(data) self.send_state = State.SEND_BODY - def send_body(self, body: bytes) -> None: + async def send_body(self, body: bytes) -> None: """ Send the request body. An empty bytes argument indicates the end of the stream: @@ -193,7 +193,7 @@ def send_body(self, body: bytes) -> None: # Transfer-Encoding: chunked self.send_seen_length += len(body) marker = f"{len(body):x}\r\n".encode("ascii") - self.stream.write(marker + body + b"\r\n") + await self.stream.write(marker + body + b"\r\n") else: # Content-Length: xxx @@ -205,13 +205,13 @@ def send_body(self, body: bytes) -> None: msg = "Not enough data sent for declared Content-Length" raise ProtocolError(msg) if body: - self.stream.write(body) + await self.stream.write(body) if body == b"": # Handle body close self.send_state = State.DONE - def recv_method_line(self) -> tuple[bytes, bytes, bytes]: + async def recv_method_line(self) -> tuple[bytes, bytes, bytes]: """ Receive the initial request method line: @@ -234,7 +234,7 @@ def recv_method_line(self) -> tuple[bytes, bytes, bytes]: # Read initial response line, eg. "GET / HTTP/1.1" exc_text = "reading request method line" - line = self.parser.read_until( + line = await self.parser.read_until( b"\r\n", max_size=4096, exc_text=exc_text, @@ -249,7 +249,7 @@ def recv_method_line(self) -> tuple[bytes, bytes, bytes]: self.send_state = State.SEND_STATUS_LINE return method, target, protocol - def recv_status_line(self) -> tuple[bytes, int, bytes]: + async def recv_status_line(self) -> tuple[bytes, int, bytes]: """ Receive the initial response status line: @@ -263,7 +263,7 @@ def recv_status_line(self) -> tuple[bytes, int, bytes]: # Read initial response line, eg. "HTTP/1.1 200 OK" exc_text = "reading response status line" - line = self.parser.read_until(b"\r\n", max_size=4096, exc_text=exc_text) + line = await self.parser.read_until(b"\r\n", max_size=4096, exc_text=exc_text) protocol, status_code_str, reason_phrase = line.split(b" ", 2) if protocol != b"HTTP/1.1": raise ProtocolError("Received unsupported protocol version") @@ -279,7 +279,7 @@ def recv_status_line(self) -> tuple[bytes, int, bytes]: self.recv_state = State.RECV_HEADERS return protocol, status_code, reason_phrase - def recv_headers(self) -> list[tuple[bytes, bytes]]: + async def recv_headers(self) -> list[tuple[bytes, bytes]]: """ Receive the response headers: @@ -295,7 +295,7 @@ def recv_headers(self) -> list[tuple[bytes, bytes]]: # Read response headers headers = [] exc_text = "reading response headers" - while line := self.parser.read_until(b"\r\n", max_size=4096, exc_text=exc_text): + while line := await self.parser.read_until(b"\r\n", max_size=4096, exc_text=exc_text): name, value = line.split(b":", 1) value = value.strip(b" ") headers.append((name, value)) @@ -326,7 +326,7 @@ def recv_headers(self) -> list[tuple[bytes, bytes]]: self.recv_state = State.RECV_BODY return headers - def recv_body(self) -> bytes: + async def recv_body(self) -> bytes: """ Receive the response body. An empty byte string indicates the end of the stream: @@ -343,26 +343,26 @@ def recv_body(self) -> bytes: if self.recv_content_length is None: # Transfer-Encoding: chunked exc_text = "reading chunk size" - line = self.parser.read_until(b"\r\n", max_size=4096, exc_text=exc_text) + line = await self.parser.read_until(b"\r\n", max_size=4096, exc_text=exc_text) sizestr, _, _ = line.partition(b";") exc_text = "Received invalid chunk size" size = bounded_hex(sizestr, max_digits=8, exc_text=exc_text) if size > 0: - body = self.parser.read(size=size) + body = await self.parser.read(size=size) exc_text = "reading chunk data" - self.parser.read_until(b"\r\n", max_size=2, exc_text=exc_text) + await self.parser.read_until(b"\r\n", max_size=2, exc_text=exc_text) self.recv_seen_length += len(body) else: body = b"" exc_text = "reading chunk termination" - self.parser.read_until(b"\r\n", max_size=2, exc_text=exc_text) + await self.parser.read_until(b"\r\n", max_size=2, exc_text=exc_text) else: # Content-Length: xxx remaining = self.recv_content_length - self.recv_seen_length size = min(remaining, 4096) - body = self.parser.read(size=size) + body = await self.parser.read(size=size) self.recv_seen_length += len(body) if self.recv_seen_length < self.recv_content_length and body == b"": msg = "Not enough data received for declared Content-Length" @@ -373,14 +373,14 @@ def recv_body(self) -> bytes: self.recv_state = State.DONE return body - def complete(self): + async def complete(self): is_fully_complete = ( self.send_state == State.DONE and self.recv_state == State.DONE ) is_keepalive = self.send_keep_alive and self.recv_keep_alive if not (is_fully_complete and is_keepalive): - self.close() + await self.close() return if self.mode == Mode.CLIENT: @@ -399,11 +399,11 @@ def complete(self): self.processing_1xx = False self.keepalive_until = time.monotonic() + self.keepalive_duration - def close(self): + async def close(self): if self.send_state != State.CLOSED: self.send_state = State.CLOSED self.recv_state = State.CLOSED - self.stream.close() + await self.stream.close() def is_idle(self) -> bool: return ( @@ -442,17 +442,17 @@ def __init__(self, stream: Stream) -> None: self._stream = stream self._chunk_size = 4096 - def _read_some(self) -> bytes: + async def _read_some(self) -> bytes: if self._buffer: ret, self._buffer = self._buffer, b"" return ret - return self._stream.read(self._chunk_size) + return await self._stream.read(self._chunk_size) def _push_back(self, buffer): assert self._buffer == b"" self._buffer = buffer - def read(self, size: int) -> bytes: + async def read(self, size: int) -> bytes: """ Read and return up to 'size' bytes from the stream, with I/O buffering provided. @@ -460,7 +460,7 @@ def read(self, size: int) -> bytes: """ buffer = bytearray() while len(buffer) < size: - chunk = self._read_some() + chunk = await self._read_some() if not chunk: break buffer.extend(chunk) @@ -470,7 +470,7 @@ def read(self, size: int) -> bytes: self._push_back(bytes(push_back)) return bytes(buffer) - def read_until( + async def read_until( self, marker: bytes, max_size: int, @@ -489,7 +489,7 @@ def read_until( """ buffer = bytearray() while len(buffer) <= max_size: - chunk = self._read_some() + chunk = await self._read_some() if not chunk: # EOF is only clean if no part of the line has arrived. if allow_empty_eof and not buffer: diff --git a/src/hip/_server.py b/src/hip/_server.py index baf09d7..b1d534f 100644 --- a/src/hip/_server.py +++ b/src/hip/_server.py @@ -25,11 +25,11 @@ def __init__(self, stream, endpoint): ) # API entry points... - def handle_requests(self): + async def handle_requests(self): try: while self._parser.is_idle(): try: - method, url, headers = self._recv_head() + method, url, headers = await self._recv_head() except EOFError: # The client closed while waiting for a request line. return @@ -39,35 +39,35 @@ def handle_requests(self): # TODO: Handle endpoint exceptions # async with Request(method, url, headers=headers, content=content) as request: try: - response = self._endpoint(request) + response = await self._endpoint(request) status_line = f"{request.method} {request.url.target} [{response.status_code}]" logger.info(status_line) except Exception: logger.exception("Internal Server Error") content = Text("Internal Server Error") err = Response(500, content=content) - self._send_head(err) - self._send_body(err) + await self._send_head(err) + await self._send_body(err) else: - self._send_head(response) - self._send_body(response) + await self._send_head(response) + await self._send_body(response) except BaseException: logger.exception("Internal Server Error") # Receive the request... - def _recv_head(self) -> tuple[str, str, list[tuple[str, str]]]: - method, target, _ = self._parser.recv_method_line() + async def _recv_head(self) -> tuple[str, str, list[tuple[str, str]]]: + method, target, _ = await self._parser.recv_method_line() m = method.decode("ascii") t = target.decode("ascii") - headers = self._parser.recv_headers() + headers = await self._parser.recv_headers() h = [(k.decode("latin-1"), v.decode("latin-1")) for k, v in headers] return m, t, h # Return the response... - def _send_head(self, response: Response): + async def _send_head(self, response: Response): protocol = b"HTTP/1.1" status, reason = response.status_code.as_tuple() - self._parser.send_status_line(protocol, status, reason) + await self._parser.send_status_line(protocol, status, reason) now = datetime.datetime.now(datetime.timezone.utc) date = now.strftime("%a, %d %b %Y, %H:%M:%S GMT") @@ -75,14 +75,14 @@ def _send_head(self, response: Response): headers = self._default_headers headers = headers.copy_set("Date", date) headers = headers.copy_update(response.headers) - self._parser.send_headers(headers.as_byte_pairs()) + await self._parser.send_headers(headers.as_byte_pairs()) - def _send_body(self, response: Response): - with response.content.open() as stream: - while data := stream.read(64 * 1024): - self._parser.send_body(data) - self._parser.send_body(b"") - self._parser.complete() + async def _send_body(self, response: Response): + async with await response.content.open() as stream: + while data := await stream.read(64 * 1024): + await self._parser.send_body(data) + await self._parser.send_body(b"") + await self._parser.complete() class Server: @@ -99,14 +99,14 @@ def __init__(self, app, host="127.0.0.1", port=8080): level=logging.DEBUG, ) - def __enter__(self): - self._tcp_server = self.backend.serve(self.host, self.port, self.handle_stream) - self._tcp_server.__enter__() + async def __aenter__(self): + self._tcp_server = await self.backend.serve(self.host, self.port, self.handle_stream) + await self._tcp_server.__enter__() logger.info(f"Serving on http://{self.host}:{self.port} (Press CTRL+C to quit)") return self - def __exit__(self, exc_type, exc_val, exc_tb): - self._tcp_server.__exit__(exc_type, exc_val, exc_tb) + async def __aexit__(self, exc_type, exc_val, exc_tb): + await self._tcp_server.__exit__(exc_type, exc_val, exc_tb) def until_interrupt(self): # pragma: no cover """Keep a context-managed server alive until Ctrl+C is pressed.""" @@ -116,6 +116,6 @@ def until_interrupt(self): # pragma: no cover except KeyboardInterrupt: pass - def handle_stream(self, stream): + async def handle_stream(self, stream): connection = HTTPConnection(stream, self.app) - connection.handle_requests() + await connection.handle_requests() diff --git a/src/hip/_streams.py b/src/hip/_streams.py index caf4b18..3d8a3b0 100644 --- a/src/hip/_streams.py +++ b/src/hip/_streams.py @@ -6,25 +6,25 @@ class Stream: - def read(self, size: int = -1) -> bytes: + async def read(self, size: int = -1) -> bytes: raise NotImplementedError() - def write(self, data: bytes) -> None: + async def write(self, data: bytes) -> None: raise NotImplementedError() - def close(self) -> None: + async def close(self) -> None: raise NotImplementedError() - def __enter__(self): + async def __aenter__(self): return self - def __exit__( + async def __aexit__( self, exc_type: type[BaseException] | None = None, exc_value: BaseException | None = None, traceback: types.TracebackType | None = None, ): - self.close() + await self.close() class ByteStream(Stream): @@ -32,10 +32,10 @@ def __init__(self, data: bytes = b""): self._buffer = io.BytesIO(data) self._size = len(data) - def read(self, size: int = -1) -> bytes: + async def read(self, size: int = -1) -> bytes: return self._buffer.read(size) - def close(self) -> None: + async def close(self) -> None: self._buffer.close() @@ -43,10 +43,10 @@ class FileStream(Stream): def __init__(self, fileobj): self._fileobj = fileobj - def read(self, size: int = -1) -> bytes: + async def read(self, size: int = -1) -> bytes: return self._fileobj.read(size) - def close(self) -> None: + async def close(self) -> None: self._fileobj.close() @@ -64,7 +64,7 @@ def __init__( self._complete = False self._buffer = io.BytesIO() - def read(self, size=-1) -> bytes: + async def read(self, size=-1) -> bytes: sections = [] length = 0 @@ -132,7 +132,7 @@ def _read_next_section(self) -> bytes: # return EOF marker return b"" - def close(self) -> None: + async def close(self) -> None: if self._fileobj is not None: self._fileobj.close() self._fileobj = None diff --git a/src/hip/backends/asyncio.py b/src/hip/backends/asyncio.py new file mode 100644 index 0000000..a853060 --- /dev/null +++ b/src/hip/backends/asyncio.py @@ -0,0 +1,83 @@ +import asyncio +import ssl +import truststore +import types + +from .._streams import Stream + + +class NetworkStream(Stream): + def __init__(self, reader: asyncio.StreamReader, writer: asyncio.StreamWriter, address: tuple[str, int]) -> None: + self._reader = reader + self._writer = writer + self._address = address + self._is_tls = False + self._is_closed = False + + @property + def host(self) -> str: + return self._address[0] + + @property + def port(self) -> int: + return self._address[1] + + async def read(self, size: int = -1) -> bytes: + if size < 0: + size = 64 * 1024 + return await self._reader.read(size) + + async def write(self, buffer: bytes) -> None: + self._writer.write(buffer) + await self._writer.drain() + + async def close(self) -> None: + if not self._is_closed: + self._is_closed = True + self._writer.close() + await self._writer.wait_closed() + + def __repr__(self): + description = "" + description += " TLS" if self._is_tls else "" + description += " CLOSED" if self._is_closed else "" + return f"" + + async def __aenter__(self) -> "Self": + return self + + async def __aexit__( + self, + exc_type: type[BaseException] | None = None, + exc_value: BaseException | None = None, + traceback: types.TracebackType | None = None, + ): + await self.close() + + +class NetworkBackend: + def __init__(self, ssl_ctx: ssl.SSLContext | None = None): + self._ssl_ctx = self.create_default_context() if ssl_ctx is None else ssl_ctx + + def create_default_context(self) -> ssl.SSLContext: + return truststore.SSLContext(ssl.PROTOCOL_TLS_CLIENT) + + async def connect(self, host: str, port: int) -> NetworkStream: + """ + Connect to the given address, returning a NetworkStream instance. + """ + address = (host, port) + reader, writer = await asyncio.open_connection(host, port) + return NetworkStream(reader, writer, address) + + async def connect_tls(self, host: str, port: int, hostname: str = "") -> NetworkStream: + """ + Connect to the given address, returning a NetworkStream instance. + """ + address = (host, port) + reader, writer = await asyncio.open_connection(host, port, ssl=self._ssl_ctx, server_hostname=hostname) + return NetworkStream(reader, writer, address) + + +Semaphore = asyncio.Semaphore +Lock = asyncio.Lock diff --git a/src/hip/transports/base.py b/src/hip/transports/base.py index a6feff2..fb15f03 100644 --- a/src/hip/transports/base.py +++ b/src/hip/transports/base.py @@ -11,27 +11,27 @@ class SendContext(tuple[StatusCode, Headers, Stream]): - def __enter__(self): + async def __aenter__(self): return self - def __exit__( + async def __aexit__( self, exc_type: type[BaseException] | None = None, exc_value: BaseException | None = None, traceback: types.TracebackType | None = None, ) -> None: _, _, stream = self - stream.close() + await stream.close() class Transport: - def send(self, request: Request) -> SendContext: + async def send(self, request: Request) -> SendContext: raise NotImplementedError() - def close(self): + async def close(self): pass - def request( + async def request( self, method: Method | str, url: URL | str, @@ -39,6 +39,6 @@ def request( content: Content | bytes | None = None, ) -> Response: request = Request(method, url, headers=headers, content=content) - with self.send(request) as (status, resp_headers, stream): - body = stream.read() + async with await self.send(request) as (status, resp_headers, stream): + body = await stream.read() return Response(status, headers=resp_headers, content=body) diff --git a/src/hip/transports/connection.py b/src/hip/transports/connection.py index 58680f6..529c399 100644 --- a/src/hip/transports/connection.py +++ b/src/hip/transports/connection.py @@ -13,7 +13,7 @@ from .._response import Response, StatusCode from .._streams import Stream from .._urls import URL -from ..backends.sync import Lock, NetworkBackend +from ..backends.asyncio import Lock, NetworkBackend from .base import SendContext, Transport __all__ = ["Connection", "open_connection"] @@ -43,21 +43,21 @@ def description(self) -> str: return self._parser.description() # API entry points... - def send(self, request: Request) -> SendContext: + async def send(self, request: Request) -> SendContext: # async with self._request_lock: # try: - self._send_head(request) - self._send_body(request) - code, headers = self._recv_head() - stream = self._recv_body(headers) + await self._send_head(request) + await self._send_body(request) + code, headers = await self._recv_head() + stream = await self._recv_body(headers) return SendContext((code, headers, stream)) - def close(self) -> None: - with self._request_lock: - self._close() + async def close(self) -> None: + async with self._request_lock: + await self._parser.close() # Top-level API for working directly with a connection. - def request( + async def request( self, method: Method | str, url: URL | str, @@ -66,61 +66,57 @@ def request( ) -> Response: url = self._origin.join(url) request = Request(method, url, headers=headers, content=content) - with self.send(request) as (status_code, resp_headers, stream): + async with await self.send(request) as (status_code, resp_headers, stream): body = stream.read() return Response(status_code, headers=resp_headers, content=body) # Send the request... - def _send_head(self, request: Request) -> None: + async def _send_head(self, request: Request) -> None: method = bytes(request.method) target = request.url.target.encode("ascii") protocol = b"HTTP/1.1" - self._parser.send_method_line(method, target, protocol) + await self._parser.send_method_line(method, target, protocol) headers = request.headers.as_byte_pairs() - self._parser.send_headers(headers) + await self._parser.send_headers(headers) - def _send_body(self, request: Request) -> None: - with request.content.open() as stream: - while data := stream.read(64 * 1024): - self._parser.send_body(data) - self._parser.send_body(b"") + async def _send_body(self, request: Request) -> None: + async with await request.content.open() as stream: + while data := await stream.read(64 * 1024): + await self._parser.send_body(data) + await self._parser.send_body(b"") # Receive the response... - def _recv_head(self) -> tuple[StatusCode, Headers]: - _, code, _ = self._parser.recv_status_line() - h = self._parser.recv_headers() + async def _recv_head(self) -> tuple[StatusCode, Headers]: + _, code, _ = await self._parser.recv_status_line() + h = await self._parser.recv_headers() status_code = StatusCode(code) headers = Headers([(k.decode("ascii"), v.decode("ascii")) for k, v in h]) return status_code, headers - def _recv_body(self, headers) -> Stream: + async def _recv_body(self, headers) -> Stream: stream: Stream = HTTPStream(self._parser, is_response=True) content_encoding = headers.get("Content-Encoding", "") if content_encoding.lower() == "gzip": stream = GZipStream(stream) return stream - # Request/response cycle complete... - def _close(self) -> None: - self._parser.close() - # Builtins... def __repr__(self) -> str: return f"" - def __enter__(self) -> "Self": + async def __aenter__(self) -> "Self": return self - def __exit__( + async def __aexit__( self, exc_type: type[BaseException] | None = None, exc_value: BaseException | None = None, traceback: types.TracebackType | None = None, ): - self.close() + await self.close() -def open_connection( +async def open_connection( url: URL | str, hostname: str = "", backend: NetworkBackend | None = None, @@ -138,8 +134,8 @@ def open_connection( port = url.port or {"http": 80, "https": 443}[url.scheme] if url.scheme == "https": - stream = backend.connect_tls(host, port, hostname) + stream = await backend.connect_tls(host, port, hostname) else: - stream = backend.connect(host, port) + stream = await backend.connect(host, port) return Connection(stream, url) diff --git a/src/hip/transports/pool.py b/src/hip/transports/pool.py index 2d2cabb..fbea52a 100644 --- a/src/hip/transports/pool.py +++ b/src/hip/transports/pool.py @@ -8,7 +8,7 @@ from .._request import Request from .._urls import URL -from ..backends.sync import NetworkBackend, Semaphore +from ..backends.asyncio import NetworkBackend, Semaphore from .base import SendContext, Transport from .connection import ( Connection, @@ -31,24 +31,24 @@ def __init__(self, backend: NetworkBackend | None = None): self._closed = False # Public API... - def send(self, request: Request) -> SendContext: + async def send(self, request: Request) -> SendContext: if self._closed: raise RuntimeError("ConnectionPool is closed.") # TODO: concurrency limiting - self._cleanup() - connection = self._get_connection(request) - return connection.send(request) + await self._cleanup() + connection = await self._get_connection(request) + return await connection.send(request) - def close(self): + async def close(self): self._closed = True closing = list(self._connections) self._connections = [] for conn in closing: - conn.close() + await conn.close() # Create or reuse connections as required... - def _get_connection(self, request: Request) -> "Connection": + async def _get_connection(self, request: Request) -> "Connection": # Attempt to reuse an existing connection. url = request.url origin = URL(scheme=url.scheme, host=url.host, port=url.port) @@ -57,18 +57,18 @@ def _get_connection(self, request: Request) -> "Connection": return conn # Or else create a new connection. - conn = open_connection( + conn = await open_connection( origin, hostname=request.headers["Host"], backend=self._network_backend ) self._connections.append(conn) return conn # Connection pool management... - def _cleanup(self) -> None: + async def _cleanup(self) -> None: time.monotonic() for conn in list(self._connections): if conn.is_expired(): - conn.close() + await conn.close() if conn.is_closed(): self._connections.remove(conn) @@ -92,13 +92,13 @@ def __del__(self): warnings.warn("ConnectionPool was garbage collected without being closed.") - def __enter__(self) -> "Self": + async def __aenter__(self) -> "Self": return self - def __exit__( + async def __aexit__( self, exc_type: type[BaseException] | None = None, exc_value: BaseException | None = None, traceback: types.TracebackType | None = None, ) -> None: - self.close() + await self.close()