Allow custom headers in multipart/form-data requests

httpx/_multipart.py

@@ -78,23 +78,39 @@ def __init__(self, name: str, value: FileTypes) -> None:
 
         fileobj: FileContent
 
+        headers: typing.Dict[str, str] = {}
+        content_type: typing.Optional[str] = None
+
         if isinstance(value, tuple):
             try:
-                filename, fileobj, content_type = value  # type: ignore
+                filename, fileobj, content_type, headers = value  # type: ignore
             except ValueError:
-                filename, fileobj = value  # type: ignore
-                content_type = guess_content_type(filename)
+                try:
+                    filename, fileobj, content_type = value  # type: ignore
+                except ValueError:
+                    filename, fileobj = value  # type: ignore
+            else:
+                headers = {k.title(): v for k, v in headers.items()}
+                if "Content-Type" in headers:
+                    raise ValueError(
+                        "Content-Type cannot be included in multipart headers"
+                    )
         else:
             filename = Path(str(getattr(value, "name", "upload"))).name
             fileobj = value
+
+        if content_type is None:
             content_type = guess_content_type(filename)
 
+        if content_type is not None:
+            headers["Content-Type"] = content_type
+
         if isinstance(fileobj, (str, io.StringIO)):
             raise TypeError(f"Expected bytes or bytes-like object got: {type(fileobj)}")
 
         self.filename = filename
         self.file = fileobj
-        self.content_type = content_type
+        self.headers = headers
         self._consumed = False
 
     def get_length(self) -> int:
@@ -122,9 +138,9 @@ def render_headers(self) -> bytes:
             if self.filename:
                 filename = format_form_param("filename", self.filename)
                 parts.extend([b"; ", filename])
-            if self.content_type is not None:
-                content_type = self.content_type.encode()
-                parts.extend([b"\r\nContent-Type: ", content_type])
+            for header_name, header_value in self.headers.items():
+                key, val = f"\r\n{header_name}: ".encode(), header_value.encode()
+                parts.extend([key, val])
             parts.append(b"\r\n\r\n")
             self._headers = b"".join(parts)
 

httpx/_types.py

@@ -89,6 +89,8 @@
     Tuple[Optional[str], FileContent],
     # (filename, file (or bytes), content_type)
     Tuple[Optional[str], FileContent, Optional[str]],
+    # (filename, file (or bytes), content_type, headers)
+    Tuple[Optional[str], FileContent, Optional[str], Mapping[str, str]],
 ]
 RequestFiles = Union[Mapping[str, FileTypes], Sequence[Tuple[str, FileTypes]]]
 

tests/test_multipart.py

@@ -94,6 +94,53 @@ def test_multipart_file_tuple():
     assert multipart["file"] == [b"<file content>"]
 
 
+@pytest.mark.parametrize("content_type", [None, "text/plain"])
+def test_multipart_file_tuple_headers(content_type: typing.Optional[str]):
+    file_name = "test.txt"
+    expected_content_type = "text/plain"
+    headers = {"Expires": "0"}
+
+    files = {"file": (file_name, io.BytesIO(b"<file content>"), content_type, headers)}
+    with mock.patch("os.urandom", return_value=os.urandom(16)):
+        boundary = os.urandom(16).hex()
+
+        headers, stream = encode_request(data={}, files=files)
+        assert isinstance(stream, typing.Iterable)
+
+        content = (
+            f'--{boundary}\r\nContent-Disposition: form-data; name="file"; '
+            f'filename="{file_name}"\r\nExpires: 0\r\nContent-Type: '
+            f"{expected_content_type}\r\n\r\n<file content>\r\n--{boundary}--\r\n"
+            "".encode("ascii")
+        )
+        assert headers == {
+            "Content-Type": f"multipart/form-data; boundary={boundary}",
+            "Content-Length": str(len(content)),
+        }
+        assert content == b"".join(stream)
+
+
+@pytest.mark.parametrize("content_type", [None, "text/plain"])
+@pytest.mark.parametrize(
+    "headers",
+    [
+        {"content-type": "text/plain"},
+        {"Content-Type": "text/plain"},
+        {"CONTENT-TYPE": "text/plain"},
+    ],
+)
+def test_multipart_headers_include_content_type(
+    content_type: typing.Optional[str], headers: typing.Dict[str, str]
+) -> None:
+    """Including contet-type in the multipart headers should not be allowed"""
+    client = httpx.Client(transport=httpx.MockTransport(echo_request_content))
+
+    files = {"file": ("test.txt", b"content", content_type, headers)}
+    pat = "Content-Type cannot be included in multipart headers"
+    with pytest.raises(ValueError, match=pat):
+        client.post("http://127.0.0.1:8000/", files=files)
+
+
 def test_multipart_encode(tmp_path: typing.Any) -> None:
     path = str(tmp_path / "name.txt")
     with open(path, "wb") as f: