diff --git a/src/hip/__init__.py b/src/hip/__init__.py index a092597..d3f77a2 100644 --- a/src/hip/__init__.py +++ b/src/hip/__init__.py @@ -1,4 +1,4 @@ -from . import backends, parsers, transports +from . import backends, parsers from .client import Client from .core.content import HTML, JSON, Binary, Content, Text from .core.forms import File, Files, Form, MultiPart @@ -35,5 +35,4 @@ "parsers", "sleep", "timeout", - "transports", ] diff --git a/src/hip/body.py b/src/hip/body.py deleted file mode 100644 index 05d9a6f..0000000 --- a/src/hip/body.py +++ /dev/null @@ -1,92 +0,0 @@ -import io -import zlib - -from .core.streams import Stream -from .parsers import HTTPParser - -__all__ = ["GZipStream", "HTTPStream"] - - -class HTTPStream(Stream): - def __init__(self, parser: HTTPParser, is_response: bool): - self._parser = parser - self._is_response = is_response - self._buffer = io.BytesIO() - self._finished = False - - def read(self, size=-1) -> bytes: - if self._finished: - return b"" - - sections = [] - length = 0 - - # If we have any data in the buffer read that and clear the buffer. - buffered = self._buffer.read() - if buffered: - sections.append(buffered) - length += len(buffered) - self._buffer.seek(0) - self._buffer.truncate(0) - - # Read each chunk in turn. - while (size < 0) or (length < size): - section = self._parser.recv_body() - sections.append(section) - length += len(section) - if section == b"": - self._finished = True - break - - # If we've more data than requested, then push some back into the buffer. - output = b"".join(sections) - if size > -1 and len(output) > size: - output, remainder = output[:size], output[size:] - self._buffer.write(remainder) - self._buffer.seek(0) - - return output - - def close(self) -> None: - self._buffer.close() - if self._is_response: - self._parser.complete() - - -class GZipStream(Stream): - def __init__(self, stream: Stream): - self._stream = stream - # Adding 16 enables gzip header and trailer handling. - self._decoder = zlib.decompressobj(wbits=16 + zlib.MAX_WBITS) - self._finished = False - self._buffer = bytearray() - - def read(self, size=-1) -> bytes: - while (size < 0 or len(self._buffer) < size) and not self._finished: - compressed = self._stream.read(8192) - - try: - if compressed: - decoded = self._decoder.decompress(compressed) - else: - decoded = self._decoder.flush() - except zlib.error as exc: - raise ValueError("Invalid gzip response content.") from exc - - if not compressed: - if not self._decoder.eof: - raise ValueError("Invalid gzip response content.") - self._finished = True - - self._buffer.extend(decoded) - if size < 0: - output = bytes(self._buffer) - self._buffer.clear() - else: - output = bytes(self._buffer[:size]) - del self._buffer[:size] - - return output - - def close(self) -> None: - self._stream.close() diff --git a/src/hip/client/__init__.py b/src/hip/client/__init__.py new file mode 100644 index 0000000..cf39c0f --- /dev/null +++ b/src/hip/client/__init__.py @@ -0,0 +1,14 @@ +from .base import Client +from .connection import Connection, GZipStream, open_connection +from .pool import ConnectionPool +from .transport import SendContext, Transport + +__all__ = [ + "Client", + "Connection", + "ConnectionPool", + "GZipStream", + "SendContext", + "Transport", + "open_connection", +] diff --git a/src/hip/client.py b/src/hip/client/base.py similarity index 94% rename from src/hip/client.py rename to src/hip/client/base.py index f71a59f..89d942f 100644 --- a/src/hip/client.py +++ b/src/hip/client/base.py @@ -1,12 +1,13 @@ import types import typing -from .core.content import HTML, JSON, Binary, Content, Text, select_media -from .core.headers import Headers -from .core.request import Method, Request -from .core.response import Response -from .core.urls import URL -from .transports import ConnectionPool, SendContext, Transport +from ..core.content import HTML, JSON, Binary, Content, Text, select_media +from ..core.headers import Headers +from ..core.request import Method, Request +from ..core.response import Response +from ..core.urls import URL +from .pool import ConnectionPool +from .transport import SendContext, Transport __all__ = ["Client"] diff --git a/src/hip/transports/connection.py b/src/hip/client/connection.py similarity index 61% rename from src/hip/transports/connection.py rename to src/hip/client/connection.py index bc5c0c2..727eb75 100644 --- a/src/hip/transports/connection.py +++ b/src/hip/client/connection.py @@ -1,3 +1,4 @@ +import io import types import typing @@ -5,8 +6,9 @@ # Once we drop 3.10 support, we can use 'typing.Self'. from typing_extensions import Self +import zlib + from ..backends.sync import Lock, NetworkBackend -from ..body import GZipStream, HTTPStream from ..core.content import Content from ..core.headers import Headers from ..core.request import Method, Request @@ -14,9 +16,9 @@ from ..core.streams import Stream from ..core.urls import URL from ..parsers import HTTPParser -from .base import SendContext, Transport +from .transport import SendContext, Transport -__all__ = ["Connection", "open_connection"] +__all__ = ["Connection", "GZipStream", "open_connection"] class Connection(Transport): @@ -94,7 +96,7 @@ def _recv_head(self) -> tuple[StatusCode, Headers]: return status_code, headers def _recv_body(self, headers) -> Stream: - stream: Stream = HTTPStream(self._parser, is_response=True) + stream: Stream = ResponseStream(self._parser) content_encoding = headers.get("Content-Encoding", "") if content_encoding.lower() == "gzip": stream = GZipStream(stream) @@ -143,3 +145,86 @@ def open_connection( stream = backend.connect(host, port) return Connection(stream, url) + + +class GZipStream(Stream): + def __init__(self, stream: Stream): + self._stream = stream + # Adding 16 enables gzip header and trailer handling. + self._decoder = zlib.decompressobj(wbits=16 + zlib.MAX_WBITS) + self._finished = False + self._buffer = bytearray() + + def read(self, size=-1) -> bytes: + while (size < 0 or len(self._buffer) < size) and not self._finished: + compressed = self._stream.read(8192) + + try: + if compressed: + decoded = self._decoder.decompress(compressed) + else: + decoded = self._decoder.flush() + except zlib.error as exc: + raise ValueError("Invalid gzip response content.") from exc + + if not compressed: + if not self._decoder.eof: + raise ValueError("Invalid gzip response content.") + self._finished = True + + self._buffer.extend(decoded) + if size < 0: + output = bytes(self._buffer) + self._buffer.clear() + else: + output = bytes(self._buffer[:size]) + del self._buffer[:size] + + return output + + def close(self) -> None: + self._stream.close() + + +class ResponseStream(Stream): + def __init__(self, parser: HTTPParser): + self._parser = parser + self._buffer = io.BytesIO() + self._finished = False + + def read(self, size=-1) -> bytes: + if self._finished: + return b"" + + sections = [] + length = 0 + + # If we have any data in the buffer read that and clear the buffer. + buffered = self._buffer.read() + if buffered: + sections.append(buffered) + length += len(buffered) + self._buffer.seek(0) + self._buffer.truncate(0) + + # Read each chunk in turn. + while (size < 0) or (length < size): + section = self._parser.recv_body() + sections.append(section) + length += len(section) + if section == b"": + self._finished = True + break + + # If we've more data than requested, then push some back into the buffer. + output = b"".join(sections) + if size > -1 and len(output) > size: + output, remainder = output[:size], output[size:] + self._buffer.write(remainder) + self._buffer.seek(0) + + return output + + def close(self) -> None: + self._buffer.close() + self._parser.complete() diff --git a/src/hip/transports/pool.py b/src/hip/client/pool.py similarity index 98% rename from src/hip/transports/pool.py rename to src/hip/client/pool.py index cc44907..035ea52 100644 --- a/src/hip/transports/pool.py +++ b/src/hip/client/pool.py @@ -9,11 +9,11 @@ from ..backends.sync import NetworkBackend, Semaphore from ..core.request import Request from ..core.urls import URL -from .base import SendContext, Transport from .connection import ( Connection, open_connection, ) +from .transport import SendContext, Transport __all__ = [ "ConnectionPool", diff --git a/src/hip/transports/base.py b/src/hip/client/transport.py similarity index 100% rename from src/hip/transports/base.py rename to src/hip/client/transport.py diff --git a/src/hip/server/__init__.py b/src/hip/server/__init__.py new file mode 100644 index 0000000..5e24a97 --- /dev/null +++ b/src/hip/server/__init__.py @@ -0,0 +1,3 @@ +from .base import Server + +__all__ = ["Server"] diff --git a/src/hip/server.py b/src/hip/server/base.py similarity index 71% rename from src/hip/server.py rename to src/hip/server/base.py index c09afec..d8b9758 100644 --- a/src/hip/server.py +++ b/src/hip/server/base.py @@ -1,14 +1,15 @@ import datetime +import io import logging -from .backends.sync import NetworkBackend -from .body import HTTPStream -from .core.content import Text -from .core.headers import Headers -from .core.request import Request -from .core.response import Response -from .core.timeouts import sleep -from .parsers import HTTPParser +from ..backends.sync import NetworkBackend +from ..core.content import Text +from ..core.headers import Headers +from ..core.request import Request +from ..core.response import Response +from ..core.streams import Stream +from ..core.timeouts import sleep +from ..parsers import HTTPParser __all__ = ["Server"] @@ -33,7 +34,7 @@ def handle_requests(self): except EOFError: # The client closed while waiting for a request line. return - with HTTPStream(self._parser, is_response=False) as stream: + with RequestStream(self._parser) as stream: body = stream.read() request = Request(method, url, headers=headers, content=body) # TODO: Handle endpoint exceptions @@ -119,3 +120,46 @@ def until_interrupt(self): # pragma: no cover def handle_stream(self, stream): connection = HTTPConnection(stream, self.app) connection.handle_requests() + + +class RequestStream(Stream): + def __init__(self, parser: HTTPParser): + self._parser = parser + self._buffer = io.BytesIO() + self._finished = False + + def read(self, size=-1) -> bytes: + if self._finished: + return b"" + + sections = [] + length = 0 + + # If we have any data in the buffer read that and clear the buffer. + buffered = self._buffer.read() + if buffered: + sections.append(buffered) + length += len(buffered) + self._buffer.seek(0) + self._buffer.truncate(0) + + # Read each chunk in turn. + while (size < 0) or (length < size): + section = self._parser.recv_body() + sections.append(section) + length += len(section) + if section == b"": + self._finished = True + break + + # If we've more data than requested, then push some back into the buffer. + output = b"".join(sections) + if size > -1 and len(output) > size: + output, remainder = output[:size], output[size:] + self._buffer.write(remainder) + self._buffer.seek(0) + + return output + + def close(self) -> None: + self._buffer.close() diff --git a/src/hip/transports/__init__.py b/src/hip/transports/__init__.py deleted file mode 100644 index a0b09a7..0000000 --- a/src/hip/transports/__init__.py +++ /dev/null @@ -1,11 +0,0 @@ -from .base import SendContext, Transport -from .connection import Connection, open_connection -from .pool import ConnectionPool - -__all__ = [ - "Connection", - "ConnectionPool", - "SendContext", - "Transport", - "open_connection", -] diff --git a/tests/test_pool.py b/tests/test_pool.py index 0dc4e93..867b586 100644 --- a/tests/test_pool.py +++ b/tests/test_pool.py @@ -15,7 +15,7 @@ def server(): def test_connection_pool_request(server): - with hip.transports.ConnectionPool() as pool: + with hip.client.ConnectionPool() as pool: assert repr(pool) == "" assert len(pool.connections) == 0 @@ -27,18 +27,18 @@ def test_connection_pool_request(server): def test_transport_request_not_implemented(): - transport = hip.transports.Transport() + transport = hip.client.Transport() with pytest.raises(NotImplementedError): transport.request("GET", "http://example.com") def test_transport_close(): - hip.transports.Transport().close() + hip.client.Transport().close() def test_connection_pool_connection_close(server): - with hip.transports.ConnectionPool() as pool: + with hip.client.ConnectionPool() as pool: assert repr(pool) == "" assert len(pool.connections) == 0 @@ -51,7 +51,7 @@ def test_connection_pool_connection_close(server): def test_connection_pool_stream(server): - with hip.transports.ConnectionPool() as pool: + with hip.client.ConnectionPool() as pool: assert repr(pool) == "" assert len(pool.connections) == 0 @@ -67,7 +67,7 @@ def test_connection_pool_stream(server): def test_connection_pool_cannot_request_after_closed(): - with hip.transports.ConnectionPool() as pool: + with hip.client.ConnectionPool() as pool: pass with pytest.raises(RuntimeError): @@ -75,13 +75,13 @@ def test_connection_pool_cannot_request_after_closed(): def test_connection_pool_should_have_managed_lifespan(server): - pool = hip.transports.ConnectionPool() + pool = hip.client.ConnectionPool() with pytest.warns(UserWarning): del pool def test_connection_request(server): - with hip.transports.open_connection(server.url) as conn: + with hip.client.open_connection(server.url) as conn: assert repr(conn) == f"" r = conn.request("GET", "/") @@ -91,7 +91,7 @@ def test_connection_request(server): def test_connection_stream(server): - with hip.transports.open_connection(server.url) as conn: + with hip.client.open_connection(server.url) as conn: assert repr(conn) == f"" request = hip.Request("GET", server.url) with conn.send(request) as (status, _, stream): @@ -102,7 +102,7 @@ def test_connection_stream(server): def test_connection_stream_read_in_chunks(server): - with hip.transports.open_connection(server.url) as conn: + with hip.client.open_connection(server.url) as conn: request = hip.Request("GET", server.url) with conn.send(request) as (status, _, stream): assert status == 200 @@ -120,7 +120,7 @@ def connect_tls(self, host, port, hostname): def test_open_https_connection(): backend = TLSBackend() - with hip.transports.open_connection( + with hip.client.open_connection( "https://example.com", hostname="example.com", backend=backend,