Tweak parsers API (#126)

Author: lovelydinosaur

docs/parsers.md

@@ -1,19 +1,22 @@
 # Parsers
 
-```python
-writer = io.BytesIO()
-reader = io.BytesIO(
-    b"HTTP/1.1 200 OK\r\n"
-    b"Content-Length: 23\r\n"
-    b"Content-Type: application/json\r\n"
-    b"\r\n"
+### Client
+
+<div class="tabs"><a onclick="httpx()" class="httpx">httpx</a> <a onclick="ahttpx()" class="ahttpx hidden">ahttpx</a></div>
+
+```{ .python .httpx }
+stream = httpx.DuplexStream(
+    b'HTTP/1.1 200 OK\r\n'
+    b'Content-Length: 23\r\n'
+    b'Content-Type: application/json\r\n'
+    b'\r\n'
     b'{"msg": "hello, world"}'
 )
-p = httpx.HTTPParser(writer, reader)
+p = ahttpx.HTTPParser(stream, mode='CLIENT')
 
 # Send the request...
-p.send_method_line(b"GET", b"/", b"HTTP/1.1")
-p.send_headers([(b"Host", b"example.com")])
+p.send_method_line(b'GET', b'/', b'HTTP/1.1')
+p.send_headers([(b'Host', b'www.example.com')])
 p.send_body(b'')
 
 # Receive the response...
@@ -24,6 +27,83 @@ while buffer := p.recv_body():
     body += buffer
 ```
 
+```{ .python .ahttpx .hidden }
+stream = ahttpx.DuplexStream(
+    b'HTTP/1.1 200 OK\r\n'
+    b'Content-Length: 23\r\n'
+    b'Content-Type: application/json\r\n'
+    b'\r\n'
+    b'{"msg": "hello, world"}'
+)
+p = ahttpx.HTTPParser(stream, mode='CLIENT')
+
+# Send the request...
+await p.send_method_line(b'GET', b'/', b'HTTP/1.1')
+await p.send_headers([(b'Host', b'www.example.com')])
+await p.send_body(b'')
+
+# Receive the response...
+protocol, code, reason_phase = await p.recv_status_line()
+headers = await p.recv_headers()
+body = b''
+while buffer := await p.recv_body():
+    body += buffer
+```
+
+### Server
+
+<div class="tabs"><a onclick="httpx()" class="httpx">httpx</a> <a onclick="ahttpx()" class="ahttpx hidden">ahttpx</a></div>
+
+```{ .python .httpx }
+stream = httpx.DuplexStream(
+    b'GET / HTTP/1.1\r\n'
+    b'Host: www.example.com\r\n'
+    b'\r\n'
+)
+p = httpx.HTTPParser(stream, mode='SERVER')
+
+# Receive the request...
+method, target, protocol = p.recv_method_line()
+headers = p.recv_headers()
+body = b''
+while buffer := p.recv_body():
+    body += buffer
+
+# Send the response...
+p.send_status_line(b'HTTP/1.1', 200, b'OK')
+p.send_headers([
+    (b'Content-Length', b'23'),
+    (b'Content-Type', b'application/json')
+])
+p.send_body(b'{"msg": "hello, world"}')
+p.send_body(b'')
+```
+
+```{ .python .ahttpx .hidden }
+stream = ahttpx.DuplexStream(
+    b'GET / HTTP/1.1\r\n'
+    b'Host: www.example.com\r\n'
+    b'\r\n'
+)
+p = ahttpx.HTTPParser(stream, mode='SERVER')
+
+# Receive the request...
+method, target, protocol = await p.recv_method_line()
+headers = await p.recv_headers()
+body = b''
+while buffer := await p.recv_body():
+    body += buffer
+
+# Send the response...
+await p.send_status_line(b'HTTP/1.1', 200, b'OK')
+await p.send_headers([
+    (b'Content-Length', b'23'),
+    (b'Content-Type', b'application/json')
+])
+await p.send_body(b'{"msg": "hello, world"}')
+await p.send_body(b'')
+```
+
 ---
 
 <span class="link-prev">← [Connections](connections.md)</span>

docs/streams.md

@@ -9,9 +9,10 @@ The interfaces here are simplified versions of Python's standard I/O operations.
 The base `Stream` class. The core of the interface is a subset of Python's `io.IOBase`...
 
 * `.read(size=-1)` - *(bytes)* Return the bytes from the data stream. If the `size` argument is omitted or negative then the entire stream will be read. If `size` is an positive integer then the call returns at most `size` bytes. A return value of `b''` indicates the end of the stream has been reached.
+* `.write(self, data: bytes)` - *None* Write the given bytes to the data stream. May raise `NotImplmentedError` if this is not a writeable stream.
 * `.close()` - Close the stream. Any further operations will raise a `ValueError`.
 
-Additionally, the following properties are also defined...
+Additionally, the following property is also defined...
 
 * `.size` - *(int or None)* Return an integer indicating the size of the stream, or `None` if the size is unknown. When working with HTTP this is used to either set a `Content-Length: <size>` header, or a `Content-Encoding: chunked` header.
 

src/ahttpx/__init__.py

@@ -3,12 +3,12 @@
 from ._content import *  # Content, File, Files, Form, HTML, JSON, MultiPart, Text
 from ._headers import *  # Headers
 from ._network import *  # NetworkBackend, NetworkStream, timeout
-from ._parsers import *  # HTTPParser, Mode, ProtocolError
+from ._parsers import *  # HTTPParser, ProtocolError
 from ._pool import *  # Connection, ConnectionPool, Transport
 from ._quickstart import *  # get, post, put, patch, delete
 from ._response import *  # Response
 from ._request import *  # Request
-from ._streams import *  # ByteStream, FileStream, HTTPStream, Stream
+from ._streams import *  # ByteStream, DuplexStream, FileStream, HTTPStream, Stream
 from ._server import *  # serve_http, run
 from ._urlencode import *  # quote, unquote, urldecode, urlencode
 from ._urls import *  # QueryParams, URL
@@ -23,6 +23,7 @@
     "ConnectionPool",
     "Content",
     "delete",
+    "DuplexStream",
     "File",
     "FileStream",
     "Files",
@@ -33,7 +34,6 @@
     "HTTPParser",
     "HTTPStream",
     "JSON",
-    "Mode",
     "MultiPart",
     "NetworkBackend",
     "NetworkStream",

src/ahttpx/_parsers.py

@@ -83,16 +83,15 @@ class HTTPParser:
     client.complete()
     client.close()
     """
-    def __init__(self, writer: Stream, reader: Stream, mode: Mode=Mode.CLIENT) -> None:
-        self.writer = writer
-        self.reader = reader
-        self.parser = ReadAheadParser(reader)
-        self.mode = mode
-
-        # Track client and server state...
-        if mode == Mode.CLIENT:
-            self.send_state = State.SEND_METHOD_LINE
-            self.recv_state = State.WAIT
+    def __init__(self, stream: Stream, mode: str) -> None:
+        self.stream = stream
+        self.parser = ReadAheadParser(stream)
+        self.mode = {'CLIENT': Mode.CLIENT, 'SERVER': Mode.SERVER}[mode]
+
+        # Track state...
+        if self.mode == Mode.CLIENT:
+            self.send_state: State = State.SEND_METHOD_LINE
+            self.recv_state: State = State.WAIT
         else:
             self.recv_state = State.RECV_METHOD_LINE
             self.send_state = State.WAIT
@@ -119,14 +118,14 @@ async def send_method_line(self, method: bytes, target: bytes, protocol: bytes)
         Sending state will switch to SEND_HEADERS state.
         """
         if self.send_state != State.SEND_METHOD_LINE:
-            msg = f"Called 'send_method_line' in invalid state {self.description()}"
+            msg = f"Called 'send_method_line' in invalid state {self.send_state}"
             raise ProtocolError(msg)
 
         # Send initial request line, eg. "GET / HTTP/1.1"
         if protocol != b'HTTP/1.1':
             raise ProtocolError("Sent unsupported protocol version")
         data = b" ".join([method, target, protocol]) + b"\r\n"
-        await self.writer.write(data)
+        await self.stream.write(data)
 
         self.send_state = State.SEND_HEADERS
         self.recv_state = State.RECV_STATUS_LINE
@@ -140,15 +139,15 @@ async def send_status_line(self, protocol: bytes, status_code: int, reason: byte
         Sending state will switch to SEND_HEADERS state.
         """
         if self.send_state != State.SEND_STATUS_LINE:
-            msg = f"Called 'send_status_line' in invalid state {self.description()}"
+            msg = f"Called 'send_status_line' in invalid state {self.send_state}"
             raise ProtocolError(msg)
 
         # Send initial request line, eg. "GET / HTTP/1.1"
         if protocol != b'HTTP/1.1':
             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"
-        await self.writer.write(data)
+        await self.stream.write(data)
 
         self.send_state = State.SEND_HEADERS
 
@@ -161,7 +160,7 @@ async def send_headers(self, headers: list[tuple[bytes, bytes]]) -> None:
         Sending state will switch to SEND_BODY state.
         """
         if self.send_state != State.SEND_HEADERS:
-            msg = f"Called 'send_headers' in invalid state {self.description()}"
+            msg = f"Called 'send_headers' in invalid state {self.send_state}"
             raise ProtocolError(msg)
 
         # Update header state
@@ -187,7 +186,7 @@ async 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"
-        await self.writer.write(data)
+        await self.stream.write(data)
 
         self.send_state = State.SEND_BODY
 
@@ -200,14 +199,14 @@ async def send_body(self, body: bytes) -> None:
         Sending state will switch to DONE.
         """
         if self.send_state != State.SEND_BODY:
-            msg = f"Called 'send_body' in invalid state {self.description()}"
+            msg = f"Called 'send_body' in invalid state {self.send_state}"
             raise ProtocolError(msg)
 
         if self.send_content_length is None:
             # Transfer-Encoding: chunked
             self.send_seen_length += len(body)
             marker = f'{len(body):x}\r\n'.encode('ascii')
-            await self.writer.write(marker + body + b'\r\n')
+            await self.stream.write(marker + body + b'\r\n')
 
         else:
             # Content-Length: xxx
@@ -219,7 +218,7 @@ async def send_body(self, body: bytes) -> None:
                 msg = 'Not enough data sent for declared Content-Length'
                 raise ProtocolError(msg)
             if body:
-                await self.writer.write(body)
+                await self.stream.write(body)
 
         if body == b'':
             # Handle body close
@@ -234,7 +233,7 @@ async def recv_method_line(self) -> tuple[bytes, bytes, bytes]:
         Receive state will switch to RECV_HEADERS.
         """
         if self.recv_state != State.RECV_METHOD_LINE:
-            msg = f"Called 'recv_method_line' in invalid state {self.description()}"
+            msg = f"Called 'recv_method_line' in invalid state {self.recv_state}"
             raise ProtocolError(msg)
 
         # Read initial response line, eg. "GET / HTTP/1.1"
@@ -257,7 +256,7 @@ async def recv_status_line(self) -> tuple[bytes, int, bytes]:
         Receive state will switch to RECV_HEADERS.
         """
         if self.recv_state != State.RECV_STATUS_LINE:
-            msg = f"Called 'recv_status_line' in invalid state {self.description()}"
+            msg = f"Called 'recv_status_line' in invalid state {self.recv_state}"
             raise ProtocolError(msg)
 
         # Read initial response line, eg. "HTTP/1.1 200 OK"
@@ -290,7 +289,7 @@ async def recv_headers(self) -> list[tuple[bytes, bytes]]:
         Receive state will revert to RECV_STATUS_CODE for interim 1xx responses.
         """
         if self.recv_state != State.RECV_HEADERS:
-            msg = f"Called 'recv_headers' in invalid state {self.description()}"
+            msg = f"Called 'recv_headers' in invalid state {self.recv_state}"
             raise ProtocolError(msg)
 
         # Read response headers
@@ -340,7 +339,7 @@ async def recv_body(self) -> bytes:
         The server will switch to DONE.
         """
         if self.recv_state != State.RECV_BODY:
-            msg = f"Called 'recv_body' in invalid state {self.description()}"
+            msg = f"Called 'recv_body' in invalid state {self.recv_state}"
             raise ProtocolError(msg)
 
         if self.recv_content_length is None:
@@ -390,6 +389,7 @@ async def complete(self):
         else:
             self.recv_state = State.RECV_METHOD_LINE
             self.send_state = State.WAIT
+
         self.send_content_length = 0
         self.recv_content_length = 0
         self.send_seen_length = 0
@@ -402,7 +402,7 @@ async def close(self):
         if self.send_state != State.CLOSED:
             self.send_state = State.CLOSED
             self.recv_state = State.CLOSED
-            await self.writer.close()
+            await self.stream.close()
 
     def is_idle(self) -> bool:
         return (

src/ahttpx/_pool.py

@@ -145,7 +145,7 @@ def __init__(self, stream: Stream, origin: URL | str):
         self._keepalive_duration = 5.0
         self._idle_expiry = time.monotonic() + self._keepalive_duration
         self._request_lock = Lock()
-        self._parser = HTTPParser(stream, stream)
+        self._parser = HTTPParser(stream, mode='CLIENT')
 
     # API for connection pool management...
     def origin(self) -> URL:

src/ahttpx/_server.py

@@ -24,33 +24,33 @@ class HTTPConnection:
     def __init__(self, stream, endpoint):
         self._stream = stream
         self._endpoint = endpoint
-        self._parser = HTTPParser(stream, stream, mode=Mode.SERVER)
+        self._parser = HTTPParser(stream, mode='SERVER')
         self._keepalive_duration = 5.0
         self._idle_expiry = time.monotonic() + self._keepalive_duration
 
     # API entry points...
     async def handle_requests(self):
-        while not self._parser.is_closed():
-            method, url, headers = await self._recv_head()
-            stream = HTTPStream(self._recv_body, self._complete)
-            # TODO: Handle endpoint exceptions
-            async with Request(method, url, headers=headers, content=stream) as request:
-                try:
-                    response = await self._endpoint(request)
-                    status_line = f"{request.method} {request.url.target} [{response.status_code} {response.reason_phrase}]"
-                    logger.info(status_line)
-                except Exception as exc:
-                    logger.error("Internal Server Error", exc_info=True)
-                    content = Text("Internal Server Error")
-                    err = Response(code=500, content=content)
-                    await self._send_head(err)
-                    await self._send_body(err)
-                else:
+        try:
+            while not self._parser.is_closed():
+                method, url, headers = await self._recv_head()
+                stream = HTTPStream(self._recv_body, self._complete)
+                # TODO: Handle endpoint exceptions
+                async with Request(method, url, headers=headers, content=stream) as request:
                     try:
-                        await self._send_head(response)
-                        await self._send_body(response)
+                        response = await self._endpoint(request)
+                        status_line = f"{request.method} {request.url.target} [{response.status_code} {response.reason_phrase}]"
+                        logger.info(status_line)
                     except Exception as exc:
                         logger.error("Internal Server Error", exc_info=True)
+                        content = Text("Internal Server Error")
+                        err = Response(code=500, content=content)
+                        await self._send_head(err)
+                        await self._send_body(err)
+                    else:
+                        await self._send_head(response)
+                        await self._send_body(response)
+        except Exception as exc:
+            logger.error("Internal Server Error", exc_info=True)
 
     async def close(self):
         self._parser.close()