allow setting an explicit multipart boundary via headers

httpx/_content.py

@@ -8,14 +8,15 @@
     Dict,
     Iterable,
     Iterator,
+    Mapping,
     Optional,
     Tuple,
     Union,
 )
 from urllib.parse import urlencode
 
 from ._exceptions import StreamClosed, StreamConsumed
-from ._multipart import MultipartStream
+from ._multipart import MultipartStream, get_multipart_boundary_from_content_type
 from ._types import (
     AsyncByteStream,
     RequestContent,
@@ -150,11 +151,21 @@ def encode_urlencoded_data(
 
 
 def encode_multipart_data(
-    data: dict, files: RequestFiles, boundary: Optional[bytes] = None
+    data: dict,
+    files: RequestFiles,
+    boundary: Optional[bytes] = None,
+    headers: Optional[Mapping[str, str]] = None,
 ) -> Tuple[Dict[str, str], MultipartStream]:
+    if headers and boundary is None and "content-type" in headers:
+        content_type = headers["content-type"]
+        if not content_type.startswith("multipart/form-data"):
+            raise ValueError(
+                f"Invalid content-type header for multipart request: {content_type}"
+            )
+        boundary = get_multipart_boundary_from_content_type(content_type)
     multipart = MultipartStream(data=data, files=files, boundary=boundary)
-    headers = multipart.get_headers()
-    return headers, multipart
+    new_headers = multipart.get_headers()
+    return new_headers, multipart
 
 
 def encode_text(text: str) -> Tuple[Dict[str, str], ByteStream]:
@@ -187,6 +198,7 @@ def encode_request(
     files: Optional[RequestFiles] = None,
     json: Optional[Any] = None,
     boundary: Optional[bytes] = None,
+    headers: Optional[Mapping[str, str]] = None,
 ) -> Tuple[Dict[str, str], Union[SyncByteStream, AsyncByteStream]]:
     """
     Handles encoding the given `content`, `data`, `files`, and `json`,
@@ -207,7 +219,7 @@ def encode_request(
     if content is not None:
         return encode_content(content)
     elif files:
-        return encode_multipart_data(data or {}, files, boundary)
+        return encode_multipart_data(data or {}, files, boundary, headers)
     elif data:
         return encode_urlencoded_data(data)
     elif json is not None:

httpx/_models.py

@@ -332,7 +332,13 @@ def __init__(
             Cookies(cookies).set_cookie_header(self)
 
         if stream is None:
-            headers, stream = encode_request(content, data, files, json)
+            headers, stream = encode_request(
+                content=content,
+                data=data,
+                files=files,
+                json=json,
+                headers=self.headers,
+            )
             self._prepare(headers)
             self.stream = stream
             # Load the request body, except for streaming content.

httpx/_multipart.py

@@ -20,6 +20,16 @@
 )
 
 
+def get_multipart_boundary_from_content_type(
+    content_type: str,
+) -> bytes:
+    if ";" in content_type:
+        for section in content_type.split(";"):
+            if section.strip().startswith("boundary="):
+                return section.strip().split("boundary=")[-1].encode("latin-1")
+    raise ValueError("Missing boundary in multipart/form-data content-type header")
+
+
 class DataField:
     """
     A single form field item, within a multipart form field.

tests/test_multipart.py

@@ -42,6 +42,52 @@ def test_multipart(value, output):
     assert multipart["file"] == [b"<file content>"]
 
 
+@pytest.mark.parametrize(
+    "header",
+    [
+        "multipart/form-data; boundary=+++; charset=utf-8",
+        "multipart/form-data; charset=utf-8; boundary=+++",
+        "multipart/form-data; boundary=+++",
+        "multipart/form-data; boundary=+++ ;",
+    ],
+)
+def test_multipart_explicit_boundary(header: str) -> None:
+    client = httpx.Client(transport=httpx.MockTransport(echo_request_content))
+
+    files = {"file": io.BytesIO(b"<file content>")}
+    headers = {"content-type": header}
+    response = client.post("http://127.0.0.1:8000/", files=files, headers=headers)
+    assert response.status_code == 200
+
+    # We're using the cgi module to verify the behavior here, which is a
+    # bit grungy, but sufficient just for our testing purposes.
+    assert response.request.headers["Content-Type"] == header
+    content_length = response.request.headers["Content-Length"]
+    pdict: dict = {
+        "boundary": b"+++",
+        "CONTENT-LENGTH": content_length,
+    }
+    multipart = cgi.parse_multipart(io.BytesIO(response.content), pdict)
+
+    assert multipart["file"] == [b"<file content>"]
+
+
+@pytest.mark.parametrize(
+    "header",
+    [
+        "multipart/form-data; charset=utf-8",
+        "multipart/form-data; charset=utf-8; ",
+    ],
+)
+def test_multipart_header_without_boundary(header: str) -> None:
+    client = httpx.Client(transport=httpx.MockTransport(echo_request_content))
+
+    files = {"file": io.BytesIO(b"<file content>")}
+    headers = {"content-type": header}
+    with pytest.raises(ValueError, match=r"Missing boundary"):
+        client.post("http://127.0.0.1:8000/", files=files, headers=headers)
+
+
 @pytest.mark.parametrize(("key"), (b"abc", 1, 2.3, None))
 def test_multipart_invalid_key(key):
     client = httpx.Client(transport=httpx.MockTransport(echo_request_content))