allow setting an explicit multipart boundary via headers

httpx/_content.py

@@ -150,11 +150,13 @@ def encode_urlencoded_data(
 
 
 def encode_multipart_data(
-    data: dict, files: RequestFiles, boundary: Optional[bytes] = None
+    data: dict,
+    files: RequestFiles,
+    boundary: Optional[bytes],
 ) -> Tuple[Dict[str, str], MultipartStream]:
     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 +189,7 @@ def encode_request(
     files: Optional[RequestFiles] = None,
     json: Optional[Any] = None,
     boundary: Optional[bytes] = None,
+    content_type: Optional[str] = None,
 ) -> Tuple[Dict[str, str], Union[SyncByteStream, AsyncByteStream]]:
     """
     Handles encoding the given `content`, `data`, `files`, and `json`,

httpx/_models.py

@@ -27,6 +27,7 @@
     StreamConsumed,
     request_context,
 )
+from ._multipart import get_multipart_boundary_from_content_type
 from ._status_codes import codes
 from ._types import (
     AsyncByteStream,
@@ -332,7 +333,15 @@ 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,
+                boundary=get_multipart_boundary_from_content_type(
+                    self.headers.get("content-type")
+                ),
+            )
             self._prepare(headers)
             self.stream = stream
             # Load the request body, except for streaming content.

httpx/_multipart.py

@@ -20,6 +20,18 @@
 )
 
 
+def get_multipart_boundary_from_content_type(
+    content_type: typing.Optional[str],
+) -> typing.Optional[bytes]:
+    if not content_type or not content_type.startswith("multipart/form-data"):
+        return None
+    if ";" in content_type:
+        for section in content_type.split(";"):
+            if section.strip().startswith("boundary="):
+                return section.strip().split("boundary=")[-1].encode("latin-1")
+    return None
+
+
 class DataField:
     """
     A single form field item, within a multipart form field.

tests/test_multipart.py

@@ -42,6 +42,54 @@ 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}
+    response = client.post("http://127.0.0.1:8000/", files=files, headers=headers)
+
+    assert response.status_code == 200
+    assert response.request.headers["Content-Type"] == header
+
+
 @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))