Streams (#73)

* Layout refactor

* Add typing

* Streams and content types

* Streams, streams, streams

* Streams, streams, streams

Author: lovelydinosaur

scripts/unasync

@@ -9,6 +9,7 @@ unasync.unasync_files(
         "src/ahttpx/_headers.py",
         "src/ahttpx/_models.py",
         "src/ahttpx/_pool.py",
+        "src/ahttpx/_streams.py",
         "src/ahttpx/_urlparse.py",
         "src/ahttpx/_urls.py"
     ],

src/ahttpx/__init__.py

@@ -4,11 +4,13 @@
 from ._models import *  # Response, Request
 from ._network import *  # NetworkBackend, NetworkStream, timeout
 from ._pool import *  # Connection, ConnectionPool, Transport, open_connection_pool, open_connection
+from ._streams import *  # ByteStream, IterByteStream, Stream
 from ._server import *  # serve_http, serve_tcp
 from ._urls import *  # QueryParams, URL
 
 
 __all__ = [
+    "ByteStream",
     "Client",
     "Connection",
     "ConnectionPool",
@@ -18,6 +20,7 @@
     "Form",
     "Headers",
     "HTML",
+    "IterByteStream",
     "JSON",
     "MultiPart",
     "NetworkBackend",
@@ -29,6 +32,7 @@
     "Request",
     "serve_http",
     "serve_tcp",
+    "Stream",
     "Text",
     "timeout",
     "Transport",

src/ahttpx/_content.py

@@ -1,9 +1,11 @@
 import json
 import os
+import re
 import pathlib
 import typing
 import urllib.parse
 
+from ._streams import Stream, ByteStream, IterByteStream
 from ._headers import Headers
 
 __all__ = [
@@ -18,13 +20,59 @@
 ]
 
 
+def parse_content_type(header: str) -> tuple[str, dict[str, str]]:
+    # The Content-Type header is described in RFC 2616 'Content-Type'
+    # https://datatracker.ietf.org/doc/html/rfc2616#section-14.17
+
+    # The 'type/subtype; parameter' format is described in RFC 2616 'Media Types'
+    # https://datatracker.ietf.org/doc/html/rfc2616#section-3.7
+
+    # Parameter quoting is described in RFC 2616 'Transfer Codings'
+    # https://datatracker.ietf.org/doc/html/rfc2616#section-3.6
+
+    header = header.strip()
+    content_type = ''
+    params = {}
+
+    # Match the content type (up to the first semicolon or end)
+    match = re.match(r'^([^;]+)', header)
+    if match:
+        content_type = match.group(1).strip().lower()
+        rest = header[match.end():]
+    else:
+        return '', {}
+
+    # Parse parameters, accounting for quoted strings
+    param_pattern = re.compile(r'''
+        ;\s*                             # Semicolon + optional whitespace
+        (?P<key>[^=;\s]+)                # Parameter key
+        =                                # Equal sign
+        (?P<value>                       # Parameter value:
+            "(?:[^"\\]|\\.)*"            #   Quoted string with escapes
+            |                            #   OR
+            [^;]*                        #   Unquoted string (until semicolon)
+        )
+    ''', re.VERBOSE)
+
+    for match in param_pattern.finditer(rest):
+        key = match.group('key').lower()
+        value = match.group('value').strip()
+        if value.startswith('"') and value.endswith('"'):
+            # Remove surrounding quotes and unescape
+            value = re.sub(r'\\(.)', r'\1', value[1:-1])
+        params[key] = value
+
+    return content_type, params
+
+
 class Content:
-    def encode(self) -> tuple[Headers, bytes | typing.AsyncIterable[bytes]]:
+    def encode(self) -> tuple[Stream, str]:
         raise NotImplementedError()
 
 
 class Form(typing.Mapping[str, str], Content):
     """
+    HTML form data, as an immutable multi-dict.
     Form parameters, as a multi-dict.
     """
 
@@ -62,6 +110,15 @@ def __init__(
 
         self._dict = d
 
+    # Content API
+
+    def encode(self) -> tuple[Stream, str]:
+        stream = ByteStream(str(self).encode("ascii"))
+        content_type = "application/x-www-form-urlencoded"
+        return (stream, content_type)
+    
+    # Dict operations
+
     def keys(self) -> typing.KeysView[str]:
         return self._dict.keys()
 
@@ -71,6 +128,13 @@ def values(self) -> typing.ValuesView[str]:
     def items(self) -> typing.ItemsView[str, str]:
         return {k: v[0] for k, v in self._dict.items()}.items()
 
+    def get(self, key: str, default: typing.Any = None) -> typing.Any:
+        if key in self._dict:
+            return self._dict[key][0]
+        return default
+
+    # Multi-dict operations
+
     def multi_items(self) -> list[tuple[str, str]]:
         multi_items: list[tuple[str, str]] = []
         for k, v in self._dict.items():
@@ -80,47 +144,27 @@ def multi_items(self) -> list[tuple[str, str]]:
     def multi_dict(self) -> dict[str, list[str]]:
         return {k: list(v) for k, v in self._dict.items()}
 
-    def get(self, key: str, default: typing.Any = None) -> typing.Any:
-        if key in self._dict:
-            return self._dict[key][0]
-        return default
-
     def get_list(self, key: str) -> list[str]:
         return list(self._dict.get(key, []))
 
-    def set(self, key: str, value: str) -> "Form":
+    # Update operations
+
+    def copy_set(self, key: str, value: str) -> "Form":
         d = self.multi_dict()
         d[key] = [value]
         return Form(d)
 
-    def append(self, key: str, value: str) -> "Form":
+    def copy_append(self, key: str, value: str) -> "Form":
         d = self.multi_dict()
         d[key] = d.get(key, []) + [value]
         return Form(d)
 
-    def remove(self, key: str) -> "Form":
-        d = {k: list(v) for k, v in self._dict.items() if k != key}
+    def copy_remove(self, key: str) -> "Form":
+        d = self.multi_dict()
+        d.pop(key, None)
         return Form(d)
 
-    def copy_with(
-        self,
-        form: (
-            typing.Mapping[str, str | typing.Sequence[str]] | typing.Sequence[tuple[str, str]] | None
-        ) = None,
-    ) -> "Form":
-        d = {k: list(v) for k, v in self._dict.items()}
-        f = Form(form)
-        return Form(dict(**d, **f._dict))
-
-    def encode(self) -> tuple[Headers, bytes | typing.AsyncIterable[bytes]]:
-        content = str(self).encode("ascii")
-        headers = Headers(
-            {
-                "Content-Type": "application/x-www-form-urlencoded",
-                "Content-Length": str(len(content)),
-            }
-        )
-        return (headers, content)
+    # Accessors & built-ins
 
     def __getitem__(self, key: str) -> str:
         return self._dict[key][0]
@@ -257,7 +301,7 @@ def get_list(self, key: str) -> list[File]:
         return list(self._dict.get(key, []))
 
     # Content interface
-    def encode(self) -> tuple[Headers, bytes | typing.AsyncIterable[bytes]]:
+    def encode(self) -> tuple[Stream, str]:
         return MultiPart(files=self).encode()
 
     # Builtins
@@ -292,32 +336,36 @@ class JSON(Content):
     def __init__(self, data: typing.Any) -> None:
         self._data = data
 
-    def encode(self) -> tuple[Headers, bytes | typing.AsyncIterable[bytes]]:
+    def encode(self) -> tuple[Stream, str]:
         content = json.dumps(
-            self._data, ensure_ascii=False, separators=(",", ":"), allow_nan=False
+            self._data,
+            ensure_ascii=False,
+            separators=(",", ":"),
+            allow_nan=False
         ).encode("utf-8")
-        headers = Headers({"Content-Type": "application/json", "Content-Length": str(len(content))})
-        return (headers, content)
+        stream = ByteStream(content)
+        content_type = "application/json"
+        return (stream, content_type)
 
 
 class Text(Content):
     def __init__(self, text: str) -> None:
         self._text = text
 
-    def encode(self) -> tuple[Headers, bytes | typing.AsyncIterable[bytes]]:
-        content = self._text.encode("utf-8")
-        headers = Headers({"Content-Type": "text/plain; charset='utf-8", "Content-Length": str(len(content))})
-        return (headers, content)
+    def encode(self) -> tuple[Stream, str]:
+        stream = ByteStream(self._text.encode("utf-8"))
+        content_type = "text/plain; charset='utf-8'"
+        return (stream, content_type)
 
 
 class HTML(Content):
     def __init__(self, text: str) -> None:
         self._text = text
 
-    def encode(self) -> tuple[Headers, bytes | typing.AsyncIterable[bytes]]:
-        content = self._text.encode("utf-8")
-        headers = Headers({"Content-Type": "text/html; charset='utf-8", "Content-Length": str(len(content))})
-        return (headers, content)
+    def encode(self) -> tuple[Stream, str]:
+        stream = ByteStream(self._text.encode("utf-8"))
+        content_type = "text/html; charset='utf-8'"
+        return (stream, content_type)
 
 
 class MultiPart(Content):
@@ -349,15 +397,10 @@ def form(self) -> Form:
     def files(self) -> Files:
         return self._files
 
-    def encode(self) -> tuple[Headers, bytes | typing.AsyncIterable[bytes]]:
-        h = Headers(
-            {
-                "Content-Type": f"multipart/form-data; boundary={self._boundary}",
-                "Transfer-Encoding": "chunked",
-            }
-        )
-        c = self.iter_bytes()
-        return (h, c)
+    def encode(self) -> tuple[Stream, str]:
+        stream = IterByteStream(self.iter_bytes())
+        content_type = f"multipart/form-data; boundary={self._boundary}"
+        return (stream, content_type)
 
     async def iter_bytes(self) -> typing.AsyncIterable[bytes]:
         for name, value in self._form.multi_items():