Merge d5c6356b87 into b5addb64f0

docs/api.md

@@ -100,7 +100,7 @@ what gets sent over the wire.*
 * `def __init__(method, url, [params], [headers], [cookies], [content], [data], [files], [json], [stream])`
 * `.method` - **str**
 * `.url` - **URL**
-* `.content` - **byte**, **byte iterator**, or **byte async iterator**
+* `.content` - **byte**, **bytearray**, **byte iterator**, or **byte async iterator**
 * `.headers` - **Headers**
 * `.cookies` - **Cookies**
 

httpx/_content.py

@@ -105,9 +105,9 @@ class UnattachedStream(AsyncByteStream, SyncByteStream):
 
 
 def encode_content(
-    content: str | bytes | Iterable[bytes] | AsyncIterable[bytes],
+    content: str | bytes | bytearray | Iterable[bytes] | AsyncIterable[bytes],
 ) -> tuple[dict[str, str], SyncByteStream | AsyncByteStream]:
-    if isinstance(content, (bytes, str)):
+    if isinstance(content, (bytes, bytearray, str)):
         body = content.encode("utf-8") if isinstance(content, str) else content
         content_length = len(body)
         headers = {"Content-Length": str(content_length)} if body else {}

httpx/_types.py

@@ -65,7 +65,7 @@ AuthTypes = Union[
     "Auth",
 ]
 
-RequestContent = Union[str, bytes, Iterable[bytes], AsyncIterable[bytes]]
+RequestContent = Union[str, bytes, bytearray, Iterable[bytes], AsyncIterable[bytes]]
 ResponseContent = Union[str, bytes, Iterable[bytes], AsyncIterable[bytes]]
 ResponseExtensions = Mapping[str, Any]
 

test

@@ -0,0 +1 @@
+# TLS secrets log file, generated by OpenSSL / Python

tests/test_content.py

@@ -50,6 +50,30 @@ async def test_bytes_content():
     assert async_content == b"Hello, world!"
 
 
+@pytest.mark.anyio
+async def test_bytearray_content():
+    request = httpx.Request(method, url, content=b"Hello, world!")
+    assert isinstance(request.stream, typing.Iterable)
+    assert isinstance(request.stream, typing.AsyncIterable)
+
+    sync_content = b"".join(list(request.stream))
+    async_content = b"".join([part async for part in request.stream])
+
+    assert request.headers == {"Host": "www.example.com", "Content-Length": "13"}
+    assert sync_content == b"Hello, world!"
+    assert async_content == b"Hello, world!"
+
+    assert isinstance(request.stream, typing.Iterable)
+    assert isinstance(request.stream, typing.AsyncIterable)
+
+    sync_content = b"".join(list(request.stream))
+    async_content = b"".join([part async for part in request.stream])
+
+    assert request.headers == {"Host": "www.example.com", "Content-Length": "13"}
+    assert sync_content == b"Hello, world!"
+    assert async_content == b"Hello, world!"
+
+
 @pytest.mark.anyio
 async def test_bytesio_content():
     request = httpx.Request(method, url, content=io.BytesIO(b"Hello, world!"))