Content vs stream (#148)

* __bytes__ interface on Content
* Add Binary to public interface
* Request/Response become properly immutable.

Author: lovelydinosaur

scripts/unasync

@@ -5,6 +5,7 @@ unasync.unasync_files(
     fpath_list = [
         "src/ahttpx/__init__.py",
         "src/ahttpx/__version__.py",
+        "src/ahttpx/_body.py",
         "src/ahttpx/_client.py",
         "src/ahttpx/_connection.py",
         "src/ahttpx/_content.py",

src/ahttpx/__init__.py

@@ -1,10 +1,10 @@
 from .__version__ import __title__, __version__
 from ._client import *  # Client
 from ._connection import *  # Connection, Transport
-from ._content import *  # Content, File, Files, Form, HTML, JSON, MultiPart, Text
+from ._content import *  # Binary, Content, File, Files, Form, HTML, JSON, MultiPart, Text
 from ._headers import *  # Headers
 from ._network import *  # NetworkBackend, NetworkStream, timeout
-from ._parsers import *  # HTTPParser, HTTPStream, ProtocolError
+from ._parsers import *  # HTTPParser, ProtocolError
 from ._pool import *  # ConnectionPool
 from ._quickstart import *  # get, post, put, patch, delete
 from ._response import *  # StatusCode, Response
@@ -18,6 +18,7 @@
 __all__ = [
     "__title__",
     "__version__",
+    "Binary",
     "ByteStream",
     "Client",
     "Connection",
@@ -33,7 +34,6 @@
     "Headers",
     "HTML",
     "HTTPParser",
-    "HTTPStream",
     "JSON",
     "Method",
     "MultiPart",

src/ahttpx/_body.py

@@ -0,0 +1,72 @@
+import io
+
+from ._content import Content
+from ._headers import Headers
+from ._parsers import HTTPParser
+from ._streams import Stream
+
+
+__all__ = ['RequestContent', 'ResponseContent', 'HTTPStream']
+
+
+class RequestContent(Content):
+    def __init__(self, parser: HTTPParser):
+        self._parser = parser
+
+    def open(self) -> Stream:
+        return HTTPStream(self._parser, is_response=False)
+
+    def headers(self) -> Headers:
+        return Headers()
+
+
+class ResponseContent(Content):
+    def __init__(self, parser: HTTPParser):
+        self._parser = parser
+
+    def open(self) -> Stream:
+        return HTTPStream(self._parser, is_response=True)
+
+    def headers(self) -> Headers:
+        return Headers()
+
+
+class HTTPStream(Stream):
+    def __init__(self, parser: HTTPParser, is_response: bool):
+        self._parser = parser
+        self._is_response = is_response
+        self._buffer = io.BytesIO()
+
+    async def read(self, size=-1) -> bytes:
+        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 = await self._parser.recv_body()
+            sections.append(section)
+            length += len(section)
+            if section == b'':
+                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
+
+    async def close(self) -> None:
+        self._buffer.close()
+        if self._is_response:
+            await self._parser.complete()

src/ahttpx/_client.py

@@ -6,7 +6,6 @@
 from ._pool import ConnectionPool, Transport
 from ._request import Method, Request
 from ._response import Response
-from ._streams import Stream
 from ._urls import URL
 
 __all__ = ["Client"]
@@ -36,7 +35,7 @@ def build_request(
         method: Method | str,
         url: URL | str,
         headers: Headers | dict[str, str] | None = None,
-        content: Content | Stream | bytes | None = None,
+        content: Content | bytes | None = None,
     ) -> Request:
         return Request(
             method=method,
@@ -50,19 +49,23 @@ async def request(
         method: Method | str,
         url: URL | str,
         headers: Headers | dict[str, str] | None = None,
-        content: Content | Stream | bytes | None = None,
+        content: Content | bytes | None = None,
     ) -> Response:
         request = self.build_request(method, url, headers=headers, content=content)
-        async with await self.via.send(request) as response:
-            await response.read()
-        return response
+        async with await self.via.send(request) as stream:
+            body = await stream.read()
+        return Response(
+            status_code=stream.status_code,
+            headers=stream.headers,
+            content=body,
+        )
 
     async def stream(
         self,
         method: Method | str,
         url: URL | str,
         headers: Headers | dict[str, str] | None = None,
-        content: Content | Stream | bytes | None = None,
+        content: Content | bytes | None = None,
     ) -> Response:
         request = self.build_request(method, url, headers=headers, content=content)
         return await self.via.send(request)
@@ -78,23 +81,23 @@ async def post(
         self,
         url: URL | str,
         headers: Headers | dict[str, str] | None = None,
-        content: Content | Stream | bytes | None = None,
+        content: Content | bytes | None = None,
     ):
         return await self.request("POST", url, headers=headers, content=content)
 
     async def put(
         self,
         url: URL | str,
         headers: Headers | dict[str, str] | None = None,
-        content: Content | Stream | bytes | None = None,
+        content: Content | bytes | None = None,
     ):
         return await self.request("PUT", url, headers=headers, content=content)
 
     async def patch(
         self,
         url: URL | str,
         headers: Headers | dict[str, str] | None = None,
-        content: Content | Stream | bytes | None = None,
+        content: Content | bytes | None = None,
     ):
         return await self.request("PATCH", url, headers=headers, content=content)
 
@@ -146,8 +149,8 @@ async def send(self, request: Request) -> Response:
             # If we have a redirect, then we read the body of the response.
             # Ensures that the HTTP connection is available for a new
             # request/response cycle.
-            await response.read()
-            await response.close()
+            async with response as stream:
+                await stream.read()
 
             # We've made a request-response and now need to issue a redirect request.
             request = self.build_redirect_request(request, response)

src/ahttpx/_connection.py

@@ -1,9 +1,10 @@
 import types
 
+from ._body import ResponseContent
 from ._content import Content
 from ._headers import Headers
 from ._network import Lock, NetworkBackend
-from ._parsers import HTTPParser, HTTPStream
+from ._parsers import HTTPParser
 from ._response import Response
 from ._request import Method, Request
 from ._streams import Stream
@@ -29,23 +30,22 @@ async def request(
         method: Method | str,
         url: URL | str,
         headers: Headers | dict[str, str] | None = None,
-        content: Content | Stream | bytes | None = None,
+        content: Content | bytes | None = None,
     ) -> Response:
         request = Request(method, url, headers=headers, content=content)
-        async with await self.send(request) as response:
-            await response.read()
-        return response
+        async with await self.send(request) as stream:
+            body = await stream.read()
+        return Response(stream.status_code, headers=stream.headers, content=body)
 
     async def stream(
         self,
         method: Method | str,
         url: URL | str,
         headers: Headers | dict[str, str] | None = None,
-        content: Content | Stream | bytes | None = None,
+        content: Content | bytes | None = None,
     ) -> Response:
         request = Request(method, url, headers=headers, content=content)
-        response = await self.send(request)
-        return response
+        return await self.send(request)
 
 
 class Connection(Transport):
@@ -78,11 +78,8 @@ async def send(self, request: Request) -> Response:
         await self._send_head(request)
         await self._send_body(request)
         code, headers = await self._recv_head()
-        stream = HTTPStream(self._parser)
-        # TODO...
-        return Response(code, headers=headers, content=stream)
-        #    finally:
-        #        await self._cycle_complete()
+        content = ResponseContent(self._parser)
+        return Response(code, headers=headers, content=content)
 
     async def close(self) -> None:
         async with self._request_lock:
@@ -94,20 +91,20 @@ async def request(
         method: Method | str,
         url: URL | str,
         headers: Headers | dict[str, str] | None = None,
-        content: Content | Stream | bytes | None = None,
+        content: Content | bytes | None = None,
     ) -> Response:
         url = self._origin.join(url)
         request = Request(method, url, headers=headers, content=content)
-        async with await self.send(request) as response:
-            await response.read()
-        return response
+        async with await self.send(request) as stream:
+            body = await stream.read()
+        return Response(stream.status_code, headers=stream.headers, content=body)
 
     async def stream(
         self,
         method: Method | str,
         url: URL | str,
         headers: Headers | dict[str, str] | None = None,
-        content: Content | Stream | bytes | None = None,
+        content: Content | bytes | None = None,
     ) -> Response:
         url = self._origin.join(url)
         request = Request(method, url, headers=headers, content=content)
@@ -123,9 +120,10 @@ async def _send_head(self, request: Request) -> None:
         await self._parser.send_headers(headers)
 
     async def _send_body(self, request: Request) -> None:
-        while data := await request.stream.read(64 * 1024):
-            await self._parser.send_body(data)
-        await self._parser.send_body(b'')
+        async with request as stream:
+            while data := await stream.read(64 * 1024):
+                await self._parser.send_body(data)
+            await self._parser.send_body(b'')
 
     # Receive the response...
     async def _recv_head(self) -> tuple[int, Headers]: