Stream interface (#118)

* Add docs for streams

* Don't expose 'tell()' interface

* Stream, ByteStream, FileStream, MultiPartStream

* Update docs

* Streams using standard File I/O interface

* Streams API

* Remove temporary file

* Context managed responses / simple Transport API

* Docs

Author: lovelydinosaur

docs/connections.md

@@ -242,6 +242,6 @@ async with await ahttpx.open_connection("http://127.0.0.1:8080") as conn:
 
 ---
 
-<span class="link-prev">← [Content Types](content-types.md)</span>
+<span class="link-prev">← [Streams](streams.md)</span>
 <span class="link-next">[Low Level Networking](networking.md) →</span>
 <span>&nbsp;</span>

docs/content-types.md

@@ -169,5 +169,5 @@ The following method must be implemented...
 ---
 
 <span class="link-prev">← [Headers](headers.md)</span>
-<span class="link-next">[Connections](connections.md) →</span>
+<span class="link-next">[Streams](streams.md) →</span>
 <span>&nbsp;</span>

docs/streams.md

@@ -0,0 +1,88 @@
+# Streams
+
+Streams provide a minimal file-like interface for reading bytes from a data source. They are used as the abstraction for reading the body of a request or response.
+
+The interfaces here are simplified versions of Python's standard I/O operations.
+
+## Stream
+
+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.
+* `.close()` - Close the stream. Any further operations will raise a `ValueError`.
+
+Additionally, the following properties are 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.
+* `.content_type` - *(str or None)* Return a string indicating the content type of the data, or `None` if the content type is unknown. When working with HTTP this is used to optionally set a `Content-Type` header.
+
+The `Stream` interface and `ContentType` interface are closely related, with streams being used as the abstraction for the bytewise representation, and content types being used to encapsulate the parsed data structure.
+
+For example, encoding some `JSON` data...
+
+```python
+>>> data = httpx.JSON({'name': 'zelda', 'score': '478'})
+>>> stream = data.encode()
+>>> stream.read()
+b'{"name":"zelda","score":"478"}'
+>>> stream.content_type
+'application/json'
+```
+
+---
+
+## ByteStream
+
+A byte stream returning fixed byte content. Similar to Python's `io.BytesIO` class.
+
+```python
+>>> s = httpx.ByteStream(b'{"msg": "Hello, world!"}')
+>>> s.read()
+b'{"msg": "Hello, world!"}'
+```
+
+## FileStream
+
+A byte stream returning content from a file.
+
+The standard pattern for instantiating a `FileStream` is to use `File` as a context manager:
+
+```python
+>>> with httpx.File('upload.json') as s:
+...     s.read()
+b'{"msg": "Hello, world!"}'
+```
+
+## MultiPartStream
+
+A byte stream returning multipart upload data.
+
+The standard pattern for instantiating a `MultiPartStream` is to use `MultiPart` as a context manager:
+
+```python
+>>> files = {'avatar-upload': 'image.png'}
+>>> with httpx.MultiPart(files=files) as s:
+...     s.read()
+# ...
+```
+
+## HTTPStream
+
+A byte stream returning unparsed content from an HTTP request or response.
+
+```python
+>>> with httpx.Client() as cli:
+...     r = cli.get('https://www.example.com/')
+...     r.stream.read()
+# ...
+```
+
+## GZipStream
+
+...
+
+---
+
+<span class="link-prev">← [Content Types](content-types.md)</span>
+<span class="link-next">[Connections](connections.md) →</span>
+<span>&nbsp;</span>

scripts/unasync

@@ -16,7 +16,7 @@ unasync.unasync_files(
         "src/ahttpx/_streams.py",
         "src/ahttpx/_urlencode.py",
         "src/ahttpx/_urlparse.py",
-        "src/ahttpx/_urls.py"
+        "src/ahttpx/_urls.py",
     ],
     rules = [
         unasync.Rule(

src/ahttpx/__init__.py

@@ -7,7 +7,7 @@
 from ._quickstart import *  # get, post, put, patch, delete
 from ._response import *  # Response
 from ._request import *  # Request
-from ._streams import *  # ByteStream, IterByteStream, FileStream, Stream
+from ._streams import *  # ByteStream, FileStream, HTTPStream, Stream
 from ._server import *  # serve_http
 from ._urlencode import *  # quote, unquote, urldecode, urlencode
 from ._urls import *  # QueryParams, URL
@@ -29,7 +29,7 @@
     "get",
     "Headers",
     "HTML",
-    "IterByteStream",
+    "HTTPStream",
     "JSON",
     "MultiPart",
     "NetworkBackend",

src/ahttpx/_client.py

@@ -1,4 +1,3 @@
-import contextlib
 import types
 import typing
 
@@ -54,21 +53,19 @@ async def request(
         content: Content | Stream | bytes | None = None,
     ) -> Response:
         request = self.build_request(method, url, headers=headers, content=content)
-        async with self.via.send(request) as response:
+        async with await self.via.send(request) as response:
             await response.read()
         return response
 
-    @contextlib.asynccontextmanager
     async def stream(
         self,
         method: str,
         url: URL | str,
         headers: Headers | typing.Mapping[str, str] | None = None,
         content: Content | Stream | bytes | None = None,
-    ) -> typing.AsyncIterator[Response]:
+    ) -> Response:
         request = self.build_request(method, url, headers=headers, content=content)
-        async with self.via.send(request) as response:
-            yield response
+        return await self.via.send(request)
 
     async def get(
         self,
@@ -139,21 +136,21 @@ def is_redirect(self, response: Response) -> bool:
     def build_redirect_request(self, request: Request, response: Response) -> Request:
         raise NotImplementedError()
 
-    @contextlib.asynccontextmanager
-    async def send(self, request: Request) -> typing.AsyncIterator[Response]:
+    async def send(self, request: Request) -> Response:
         while True:
-            async with self._transport.send(request) as response:
-                if not self.is_redirect(response):
-                    yield response
-                    return
+            response = await self._transport.send(request)
+
+            if not self.is_redirect(response):
+                return 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()
+            # 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()
 
             # We've made a request-response and now need to issue a redirect request.
             request = self.build_redirect_request(request, response)
 
-    async def aclose(self):
+    async def close(self):
         pass

src/ahttpx/_content.py

@@ -2,7 +2,7 @@
 import os
 import typing
 
-from ._streams import Stream, ByteStream, FileStream, IterByteStream
+from ._streams import Stream, ByteStream, FileStream, MultiPartStream
 from ._urlencode import urldecode, urlencode
 
 __all__ = [
@@ -16,21 +16,9 @@
     "HTML",
 ]
 
-# https://github.com/nginx/nginx/blob/master/conf/mime.types
-_content_types = {
-    ".json": "application/json",
-    ".js": "application/javascript",
-    ".html": "text/html",
-    ".css": "text/css",
-    ".png": "image/png",
-    ".jpeg": "image/jpeg",
-    ".jpg": "image/jpeg",
-    ".gif": "image/gif",
-}
-
 
 class Content:
-    def encode(self) -> tuple[Stream, str]:
+    def encode(self) -> Stream:
         raise NotImplementedError()
 
 
@@ -73,11 +61,11 @@ def __init__(
 
     # Content API
 
-    def encode(self) -> tuple[Stream, str]:
-        stream = ByteStream(str(self).encode("ascii"))
+    def encode(self) -> Stream:
+        content = str(self).encode("ascii")
         content_type = "application/x-www-form-urlencoded"
-        return (stream, content_type)
-    
+        return ByteStream(content, content_type)
+
     # Dict operations
 
     def keys(self) -> typing.KeysView[str]:
@@ -172,17 +160,8 @@ def name(self) -> str:
     def size(self) -> int:
         return os.path.getsize(self._path)
 
-    def content_type(self) -> str:
-        _, ext = os.path.splitext(self._path)
-        ct = _content_types.get(ext, "application/octet-stream")
-        if ct.startswith('text/'):
-            ct += "; charset='utf-8'"
-        return ct
-
-    def encode(self) -> tuple[Stream, str]:
-        stream = FileStream(self._path)
-        content_type = self.content_type()
-        return (stream, content_type)
+    def encode(self) -> Stream:
+        return FileStream(self._path)
 
     def __lt__(self, other: typing.Any) -> bool:
         return isinstance(other, File) and other._path < self._path
@@ -249,7 +228,7 @@ def get_list(self, key: str) -> list[File]:
         return list(self._dict.get(key, []))
 
     # Content interface
-    def encode(self) -> tuple[Stream, str]:
+    def encode(self) -> Stream:
         return MultiPart(files=self).encode()
 
     # Builtins
@@ -282,16 +261,15 @@ class JSON(Content):
     def __init__(self, data: typing.Any) -> None:
         self._data = data
 
-    def encode(self) -> tuple[Stream, str]:
+    def encode(self) -> Stream:
         content = json.dumps(
             self._data,
             ensure_ascii=False,
             separators=(",", ":"),
             allow_nan=False
         ).encode("utf-8")
-        stream = ByteStream(content)
         content_type = "application/json"
-        return (stream, content_type)
+        return ByteStream(content, content_type)
 
     def __repr__(self) -> str:
         return f"<JSON {self._data!r}>"
@@ -301,10 +279,10 @@ class Text(Content):
     def __init__(self, text: str) -> None:
         self._text = text
 
-    def encode(self) -> tuple[Stream, str]:
-        stream = ByteStream(self._text.encode("utf-8"))
+    def encode(self) -> Stream:
+        content = self._text.encode("utf-8")
         content_type = "text/plain; charset='utf-8'"
-        return (stream, content_type)
+        return ByteStream(content, content_type)
 
     def __repr__(self) -> str:
         return f"<Text {self._text!r}>"
@@ -314,10 +292,10 @@ class HTML(Content):
     def __init__(self, text: str) -> None:
         self._text = text
 
-    def encode(self) -> tuple[Stream, str]:
-        stream = ByteStream(self._text.encode("utf-8"))
+    def encode(self) -> Stream:
+        content = self._text.encode("utf-8")
         content_type = "text/html; charset='utf-8'"
-        return (stream, content_type)
+        return ByteStream(content, content_type)
 
     def __repr__(self) -> str:
         return f"<HTML {self._text!r}>"
@@ -353,37 +331,10 @@ def form(self) -> Form:
     def files(self) -> Files:
         return self._files
 
-    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.AsyncIterator[bytes]:
-        for key, value in self._form.multi_items():
-            # See https://html.spec.whatwg.org/ - LF, CR, and " must be percent escaped.
-            name = key.translate({10: "%0A", 13: "%0D", 34: "%22"})
-            yield (
-                f"--{self._boundary}\r\n"
-                f'Content-Disposition: form-data; name="{name}"\r\n'
-                f"\r\n"
-                f"{value}\r\n"
-            ).encode("utf-8")
-
-        for key, file in self._files.multi_items():
-            # See https://html.spec.whatwg.org/ - LF, CR, and " must be percent escaped.
-            name = key.translate({10: "%0A", 13: "%0D", 34: "%22"})
-            filename = file.name().translate({10: "%0A", 13: "%0D", 34: "%22"})
-            stream, _ = file.encode()
-            yield (
-                f"--{self._boundary}\r\n"
-                f'Content-Disposition: form-data; name="{name}"; filename="{filename}"\r\n'
-                f"\r\n"
-            ).encode("utf-8")
-            async for buffer in stream:
-                yield buffer
-            yield "\r\n".encode("utf-8")
-
-        yield f"--{self._boundary}--\r\n".encode("utf-8")
+    def encode(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)
 
     def __repr__(self) -> str:
         return f"<MultiPart form={self._form.multi_items()!r}, files={self._files.multi_items()!r}>"

src/ahttpx/_headers.py

@@ -23,14 +23,14 @@
 
 def headername(name: str) -> str:
     if name.strip(VALID_HEADER_CHARS) or not name:
-        raise ValueError("Invalid HTTP header name {key!r}.")
+        raise ValueError(f"Invalid HTTP header name {name!r}.")
     return name
 
 
 def headervalue(value: str) -> str:
     value = value.strip(" ")
     if not value or not value.isascii() or not value.isprintable():
-        raise ValueError("Invalid HTTP header value {key!r}.")
+        raise ValueError(f"Invalid HTTP header value {value!r}.")
     return value
 
 

src/ahttpx/_network.py

@@ -31,9 +31,10 @@ async def start_tls(self, ctx: ssl.SSLContext, hostname: str | None = None) -> N
         self._tls = True
 
     async def close(self) -> None:
-        self._writer.close()
-        await self._writer.wait_closed()
-        self._closed = True
+        if not self._closed:
+            self._writer.close()
+            await self._writer.wait_closed()
+            self._closed = True
 
     def __repr__(self):
         description = ""