We are currently solving the following issue within our repository. Here is the issue text:
--- BEGIN ISSUE ---
Enforce that upload files must be binary, not text.
### Based on this discussion... https://github.com/encode/httpx/discussions/1704

<div type='discussions-op-text'>

<sup>Originally posted by **plotski** June 21, 2021</sup>
I just spent several hours debugging `LocalProtocolError("Too much data for declared Content-Length")`. I've figured out what's going on, but I don't know what the best fix would be or if this is entirely my own fault.

I was passing a `StringIO` object to a POST request, e.g.:

```python
>>> files = {'upload-file': ('foo', StringIO('cöntént of föö'))}
>>> r = httpx.post("https://httpbin.org/post", files=files)
```

The `StringIO` object ends up in `FileField.get_length()`, which calls `peek_filelike_length()`, which correctly reports the the length of the string. However, we must know the length of the string as `bytes` because that's what we need for the POST request.

The fix is very simple: Use `BytesIO` instead of `StringIO`. But the error message was extremely unhelpful.

Maybe `FileField` shouldn't accept a `StringIO` object? Or maybe it should convert to `bytes` first? Or maybe this is really obvious for anyone with more experience and there's no need to fix it?</div>

I'm fairly sure that the most sensible thing to do here is to raise an error on:

* Files opened in text mode.
* `StringIO` instances, used instead of `BytesIO` instances.

--- END ISSUE ---

Below are some code segments, each from a relevant file. One or more of these files may contain bugs.
--- BEGIN FILES ---
Path: `httpx/_types.py`
Content:
```
1 """
2 Type definitions for type checking purposes.
3 """
4 
5 import ssl
6 from http.cookiejar import CookieJar
7 from typing import (
8     IO,
9     TYPE_CHECKING,
10     AsyncIterable,
11     Callable,
12     Dict,
13     Iterable,
14     List,
15     Mapping,
16     Optional,
17     Sequence,
18     Tuple,
19     Union,
20 )
21 
22 if TYPE_CHECKING:  # pragma: no cover
23     from ._auth import Auth  # noqa: F401
24     from ._config import Proxy, Timeout  # noqa: F401
25     from ._models import URL, Cookies, Headers, QueryParams, Request  # noqa: F401
26 
27 
28 PrimitiveData = Optional[Union[str, int, float, bool]]
29 
30 RawURL = Tuple[bytes, bytes, Optional[int], bytes]
31 
32 URLTypes = Union["URL", str]
33 
34 QueryParamTypes = Union[
35     "QueryParams",
36     Mapping[str, Union[PrimitiveData, Sequence[PrimitiveData]]],
37     List[Tuple[str, PrimitiveData]],
38     Tuple[Tuple[str, PrimitiveData], ...],
39     str,
40     bytes,
41     None,
42 ]
43 
44 HeaderTypes = Union[
45     "Headers",
46     Dict[str, str],
47     Dict[bytes, bytes],
48     Sequence[Tuple[str, str]],
49     Sequence[Tuple[bytes, bytes]],
50 ]
51 
52 CookieTypes = Union["Cookies", CookieJar, Dict[str, str], List[Tuple[str, str]]]
53 
54 CertTypes = Union[
55     # certfile
56     str,
57     # (certfile, keyfile)
58     Tuple[str, Optional[str]],
59     # (certfile, keyfile, password)
60     Tuple[str, Optional[str], Optional[str]],
61 ]
62 VerifyTypes = Union[str, bool, ssl.SSLContext]
63 TimeoutTypes = Union[
64     Optional[float],
65     Tuple[Optional[float], Optional[float], Optional[float], Optional[float]],
66     "Timeout",
67 ]
68 ProxiesTypes = Union[URLTypes, "Proxy", Dict[URLTypes, Union[None, URLTypes, "Proxy"]]]
69 
70 AuthTypes = Union[
71     Tuple[Union[str, bytes], Union[str, bytes]],
72     Callable[["Request"], "Request"],
73     "Auth",
74     None,
75 ]
76 
77 RequestContent = Union[str, bytes, Iterable[bytes], AsyncIterable[bytes]]
78 ResponseContent = Union[str, bytes, Iterable[bytes], AsyncIterable[bytes]]
79 
80 RequestData = dict
81 
82 FileContent = Union[IO[str], IO[bytes], str, bytes]
83 FileTypes = Union[
84     # file (or text)
85     FileContent,
86     # (filename, file (or text))
87     Tuple[Optional[str], FileContent],
88     # (filename, file (or text), content_type)
89     Tuple[Optional[str], FileContent, Optional[str]],
90 ]
91 RequestFiles = Union[Mapping[str, FileTypes], Sequence[Tuple[str, FileTypes]]]
92
```

Path: `httpx/_multipart.py`
Content:
```
1 import binascii
2 import os
3 import typing
4 from pathlib import Path
5 
6 from ._transports.base import AsyncByteStream, SyncByteStream
7 from ._types import FileContent, FileTypes, RequestFiles
8 from ._utils import (
9     format_form_param,
10     guess_content_type,
11     peek_filelike_length,
12     primitive_value_to_str,
13     to_bytes,
14 )
15 
16 
17 class DataField:
18     """
19     A single form field item, within a multipart form field.
20     """
21 
22     def __init__(
23         self, name: str, value: typing.Union[str, bytes, int, float, None]
24     ) -> None:
25         if not isinstance(name, str):
26             raise TypeError(
27                 f"Invalid type for name. Expected str, got {type(name)}: {name!r}"
28             )
29         if value is not None and not isinstance(value, (str, bytes, int, float)):
30             raise TypeError(
31                 f"Invalid type for value. Expected primitive type, got {type(value)}: {value!r}"
32             )
33         self.name = name
34         self.value: typing.Union[str, bytes] = (
35             value if isinstance(value, bytes) else primitive_value_to_str(value)
36         )
37 
38     def render_headers(self) -> bytes:
39         if not hasattr(self, "_headers"):
40             name = format_form_param("name", self.name)
41             self._headers = b"".join(
42                 [b"Content-Disposition: form-data; ", name, b"\r\n\r\n"]
43             )
44 
45         return self._headers
46 
47     def render_data(self) -> bytes:
48         if not hasattr(self, "_data"):
49             self._data = to_bytes(self.value)
50 
51         return self._data
52 
53     def get_length(self) -> int:
54         headers = self.render_headers()
55         data = self.render_data()
56         return len(headers) + len(data)
57 
58     def render(self) -> typing.Iterator[bytes]:
59         yield self.render_headers()
60         yield self.render_data()
61 
62 
63 class FileField:
64     """
65     A single file field item, within a multipart form field.
66     """
67 
68     def __init__(self, name: str, value: FileTypes) -> None:
69         self.name = name
70 
71         fileobj: FileContent
72 
73         if isinstance(value, tuple):
74             try:
75                 filename, fileobj, content_type = value  # type: ignore
76             except ValueError:
77                 filename, fileobj = value  # type: ignore
78                 content_type = guess_content_type(filename)
79         else:
80             filename = Path(str(getattr(value, "name", "upload"))).name
81             fileobj = value
82             content_type = guess_content_type(filename)
83 
84         self.filename = filename
85         self.file = fileobj
86         self.content_type = content_type
87         self._consumed = False
88 
89     def get_length(self) -> int:
90         headers = self.render_headers()
91 
92         if isinstance(self.file, (str, bytes)):
93             return len(headers) + len(to_bytes(self.file))
94 
95         # Let's do our best not to read `file` into memory.
96         file_length = peek_filelike_length(self.file)
97         if file_length is None:
98             # As a last resort, read file and cache contents for later.
99             assert not hasattr(self, "_data")
100             self._data = to_bytes(self.file.read())
101             file_length = len(self._data)
102 
103         return len(headers) + file_length
104 
105     def render_headers(self) -> bytes:
106         if not hasattr(self, "_headers"):
107             parts = [
108                 b"Content-Disposition: form-data; ",
109                 format_form_param("name", self.name),
110             ]
111             if self.filename:
112                 filename = format_form_param("filename", self.filename)
113                 parts.extend([b"; ", filename])
114             if self.content_type is not None:
115                 content_type = self.content_type.encode()
116                 parts.extend([b"\r\nContent-Type: ", content_type])
117             parts.append(b"\r\n\r\n")
118             self._headers = b"".join(parts)
119 
120         return self._headers
121 
122     def render_data(self) -> typing.Iterator[bytes]:
123         if isinstance(self.file, (str, bytes)):
124             yield to_bytes(self.file)
125             return
126 
127         if hasattr(self, "_data"):
128             # Already rendered.
129             yield self._data
130             return
131 
132         if self._consumed:  # pragma: nocover
133             self.file.seek(0)
134         self._consumed = True
135 
136         for chunk in self.file:
137             yield to_bytes(chunk)
138 
139     def render(self) -> typing.Iterator[bytes]:
140         yield self.render_headers()
141         yield from self.render_data()
142 
143 
144 class MultipartStream(SyncByteStream, AsyncByteStream):
145     """
146     Request content as streaming multipart encoded form data.
147     """
148 
149     def __init__(self, data: dict, files: RequestFiles, boundary: bytes = None) -> None:
150         if boundary is None:
151             boundary = binascii.hexlify(os.urandom(16))
152 
153         self.boundary = boundary
154         self.content_type = "multipart/form-data; boundary=%s" % boundary.decode(
155             "ascii"
156         )
157         self.fields = list(self._iter_fields(data, files))
158 
159     def _iter_fields(
160         self, data: dict, files: RequestFiles
161     ) -> typing.Iterator[typing.Union[FileField, DataField]]:
162         for name, value in data.items():
163             if isinstance(value, list):
164                 for item in value:
165                     yield DataField(name=name, value=item)
166             else:
167                 yield DataField(name=name, value=value)
168 
169         file_items = files.items() if isinstance(files, typing.Mapping) else files
170         for name, value in file_items:
171             yield FileField(name=name, value=value)
172 
173     def iter_chunks(self) -> typing.Iterator[bytes]:
174         for field in self.fields:
175             yield b"--%s\r\n" % self.boundary
176             yield from field.render()
177             yield b"\r\n"
178         yield b"--%s--\r\n" % self.boundary
179 
180     def iter_chunks_lengths(self) -> typing.Iterator[int]:
181         boundary_length = len(self.boundary)
182         # Follow closely what `.iter_chunks()` does.
183         for field in self.fields:
184             yield 2 + boundary_length + 2
185             yield field.get_length()
186             yield 2
187         yield 2 + boundary_length + 4
188 
189     def get_content_length(self) -> int:
190         return sum(self.iter_chunks_lengths())
191 
192     # Content stream interface.
193 
194     def get_headers(self) -> typing.Dict[str, str]:
195         content_length = str(self.get_content_length())
196         content_type = self.content_type
197         return {"Content-Length": content_length, "Content-Type": content_type}
198 
199     def __iter__(self) -> typing.Iterator[bytes]:
200         for chunk in self.iter_chunks():
201             yield chunk
202 
203     async def __aiter__(self) -> typing.AsyncIterator[bytes]:
204         for chunk in self.iter_chunks():
205             yield chunk
206
```
--- END FILES ---

Please first localize the bug based on the issue statement, and then generate a patch according to the `git diff` format fenced by three backticks.

Here is an example:

```diff
diff --git a/examples/server_async.py b/examples/server_async.py
--- a/examples/server_async.py
+++ b/examples/server_async.py
@@ -313,4 +313,4 @@
 
 
 if __name__ == "__main__":
-    asyncio.run(run_async_server("."), debug=True)
+    asyncio.run(run_async_server(), debug=True)
diff --git a/examples/server_sync.py b/examples/server_sync.py
--- a/examples/server_sync.py
+++ b/examples/server_sync.py
@@ -313,5 +313,5 @@
 
 
 if __name__ == "__main__":
-    server = run_sync_server(".")
+    server = run_sync_server()
     server.shutdown()

```

diff --git a/httpx/_multipart.py b/httpx/_multipart.py
--- a/httpx/_multipart.py
+++ b/httpx/_multipart.py
@@ -1,4 +1,5 @@
 import binascii
+import io
 import os
 import typing
 from pathlib import Path
@@ -81,6 +82,9 @@
             fileobj = value
             content_type = guess_content_type(filename)
 
+        if isinstance(fileobj, str) or isinstance(fileobj, 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
diff --git a/httpx/_types.py b/httpx/_types.py
--- a/httpx/_types.py
+++ b/httpx/_types.py
@@ -79,7 +79,7 @@
 
 RequestData = dict
 
-FileContent = Union[IO[str], IO[bytes], str, bytes]
+FileContent = Union[IO[bytes], bytes]
 FileTypes = Union[
     # file (or text)
     FileContent,

{"golden_diff": "diff --git a/httpx/_multipart.py b/httpx/_multipart.py\n--- a/httpx/_multipart.py\n+++ b/httpx/_multipart.py\n@@ -1,4 +1,5 @@\n import binascii\n+import io\n import os\n import typing\n from pathlib import Path\n@@ -81,6 +82,9 @@\n             fileobj = value\n             content_type = guess_content_type(filename)\n \n+        if isinstance(fileobj, str) or isinstance(fileobj, io.StringIO):\n+            raise TypeError(f\"Expected bytes or bytes-like object got: {type(fileobj)}\")\n+\n         self.filename = filename\n         self.file = fileobj\n         self.content_type = content_type\ndiff --git a/httpx/_types.py b/httpx/_types.py\n--- a/httpx/_types.py\n+++ b/httpx/_types.py\n@@ -79,7 +79,7 @@\n \n RequestData = dict\n \n-FileContent = Union[IO[str], IO[bytes], str, bytes]\n+FileContent = Union[IO[bytes], bytes]\n FileTypes = Union[\n     # file (or text)\n     FileContent,\n", "issue": "Enforce that upload files must be binary, not text.\n### Based on this discussion... https://github.com/encode/httpx/discussions/1704\r\n\r\n<div type='discussions-op-text'>\r\n\r\n<sup>Originally posted by **plotski** June 21, 2021</sup>\r\nI just spent several hours debugging `LocalProtocolError(\"Too much data for declared Content-Length\")`. I've figured out what's going on, but I don't know what the best fix would be or if this is entirely my own fault.\r\n\r\nI was passing a `StringIO` object to a POST request, e.g.:\r\n\r\n```python\r\n>>> files = {'upload-file': ('foo', StringIO('c\u00f6nt\u00e9nt of f\u00f6\u00f6'))}\r\n>>> r = httpx.post(\"https://httpbin.org/post\", files=files)\r\n```\r\n\r\nThe `StringIO` object ends up in `FileField.get_length()`, which calls `peek_filelike_length()`, which correctly reports the the length of the string. However, we must know the length of the string as `bytes` because that's what we need for the POST request.\r\n\r\nThe fix is very simple: Use `BytesIO` instead of `StringIO`. But the error message was extremely unhelpful.\r\n\r\nMaybe `FileField` shouldn't accept a `StringIO` object? Or maybe it should convert to `bytes` first? Or maybe this is really obvious for anyone with more experience and there's no need to fix it?</div>\r\n\r\nI'm fairly sure that the most sensible thing to do here is to raise an error on:\r\n\r\n* Files opened in text mode.\r\n* `StringIO` instances, used instead of `BytesIO` instances.\n", "before_files": [{"content": "\"\"\"\nType definitions for type checking purposes.\n\"\"\"\n\nimport ssl\nfrom http.cookiejar import CookieJar\nfrom typing import (\n    IO,\n    TYPE_CHECKING,\n    AsyncIterable,\n    Callable,\n    Dict,\n    Iterable,\n    List,\n    Mapping,\n    Optional,\n    Sequence,\n    Tuple,\n    Union,\n)\n\nif TYPE_CHECKING:  # pragma: no cover\n    from ._auth import Auth  # noqa: F401\n    from ._config import Proxy, Timeout  # noqa: F401\n    from ._models import URL, Cookies, Headers, QueryParams, Request  # noqa: F401\n\n\nPrimitiveData = Optional[Union[str, int, float, bool]]\n\nRawURL = Tuple[bytes, bytes, Optional[int], bytes]\n\nURLTypes = Union[\"URL\", str]\n\nQueryParamTypes = Union[\n    \"QueryParams\",\n    Mapping[str, Union[PrimitiveData, Sequence[PrimitiveData]]],\n    List[Tuple[str, PrimitiveData]],\n    Tuple[Tuple[str, PrimitiveData], ...],\n    str,\n    bytes,\n    None,\n]\n\nHeaderTypes = Union[\n    \"Headers\",\n    Dict[str, str],\n    Dict[bytes, bytes],\n    Sequence[Tuple[str, str]],\n    Sequence[Tuple[bytes, bytes]],\n]\n\nCookieTypes = Union[\"Cookies\", CookieJar, Dict[str, str], List[Tuple[str, str]]]\n\nCertTypes = Union[\n    # certfile\n    str,\n    # (certfile, keyfile)\n    Tuple[str, Optional[str]],\n    # (certfile, keyfile, password)\n    Tuple[str, Optional[str], Optional[str]],\n]\nVerifyTypes = Union[str, bool, ssl.SSLContext]\nTimeoutTypes = Union[\n    Optional[float],\n    Tuple[Optional[float], Optional[float], Optional[float], Optional[float]],\n    \"Timeout\",\n]\nProxiesTypes = Union[URLTypes, \"Proxy\", Dict[URLTypes, Union[None, URLTypes, \"Proxy\"]]]\n\nAuthTypes = Union[\n    Tuple[Union[str, bytes], Union[str, bytes]],\n    Callable[[\"Request\"], \"Request\"],\n    \"Auth\",\n    None,\n]\n\nRequestContent = Union[str, bytes, Iterable[bytes], AsyncIterable[bytes]]\nResponseContent = Union[str, bytes, Iterable[bytes], AsyncIterable[bytes]]\n\nRequestData = dict\n\nFileContent = Union[IO[str], IO[bytes], str, bytes]\nFileTypes = Union[\n    # file (or text)\n    FileContent,\n    # (filename, file (or text))\n    Tuple[Optional[str], FileContent],\n    # (filename, file (or text), content_type)\n    Tuple[Optional[str], FileContent, Optional[str]],\n]\nRequestFiles = Union[Mapping[str, FileTypes], Sequence[Tuple[str, FileTypes]]]\n", "path": "httpx/_types.py"}, {"content": "import binascii\nimport os\nimport typing\nfrom pathlib import Path\n\nfrom ._transports.base import AsyncByteStream, SyncByteStream\nfrom ._types import FileContent, FileTypes, RequestFiles\nfrom ._utils import (\n    format_form_param,\n    guess_content_type,\n    peek_filelike_length,\n    primitive_value_to_str,\n    to_bytes,\n)\n\n\nclass DataField:\n    \"\"\"\n    A single form field item, within a multipart form field.\n    \"\"\"\n\n    def __init__(\n        self, name: str, value: typing.Union[str, bytes, int, float, None]\n    ) -> None:\n        if not isinstance(name, str):\n            raise TypeError(\n                f\"Invalid type for name. Expected str, got {type(name)}: {name!r}\"\n            )\n        if value is not None and not isinstance(value, (str, bytes, int, float)):\n            raise TypeError(\n                f\"Invalid type for value. Expected primitive type, got {type(value)}: {value!r}\"\n            )\n        self.name = name\n        self.value: typing.Union[str, bytes] = (\n            value if isinstance(value, bytes) else primitive_value_to_str(value)\n        )\n\n    def render_headers(self) -> bytes:\n        if not hasattr(self, \"_headers\"):\n            name = format_form_param(\"name\", self.name)\n            self._headers = b\"\".join(\n                [b\"Content-Disposition: form-data; \", name, b\"\\r\\n\\r\\n\"]\n            )\n\n        return self._headers\n\n    def render_data(self) -> bytes:\n        if not hasattr(self, \"_data\"):\n            self._data = to_bytes(self.value)\n\n        return self._data\n\n    def get_length(self) -> int:\n        headers = self.render_headers()\n        data = self.render_data()\n        return len(headers) + len(data)\n\n    def render(self) -> typing.Iterator[bytes]:\n        yield self.render_headers()\n        yield self.render_data()\n\n\nclass FileField:\n    \"\"\"\n    A single file field item, within a multipart form field.\n    \"\"\"\n\n    def __init__(self, name: str, value: FileTypes) -> None:\n        self.name = name\n\n        fileobj: FileContent\n\n        if isinstance(value, tuple):\n            try:\n                filename, fileobj, content_type = value  # type: ignore\n            except ValueError:\n                filename, fileobj = value  # type: ignore\n                content_type = guess_content_type(filename)\n        else:\n            filename = Path(str(getattr(value, \"name\", \"upload\"))).name\n            fileobj = value\n            content_type = guess_content_type(filename)\n\n        self.filename = filename\n        self.file = fileobj\n        self.content_type = content_type\n        self._consumed = False\n\n    def get_length(self) -> int:\n        headers = self.render_headers()\n\n        if isinstance(self.file, (str, bytes)):\n            return len(headers) + len(to_bytes(self.file))\n\n        # Let's do our best not to read `file` into memory.\n        file_length = peek_filelike_length(self.file)\n        if file_length is None:\n            # As a last resort, read file and cache contents for later.\n            assert not hasattr(self, \"_data\")\n            self._data = to_bytes(self.file.read())\n            file_length = len(self._data)\n\n        return len(headers) + file_length\n\n    def render_headers(self) -> bytes:\n        if not hasattr(self, \"_headers\"):\n            parts = [\n                b\"Content-Disposition: form-data; \",\n                format_form_param(\"name\", self.name),\n            ]\n            if self.filename:\n                filename = format_form_param(\"filename\", self.filename)\n                parts.extend([b\"; \", filename])\n            if self.content_type is not None:\n                content_type = self.content_type.encode()\n                parts.extend([b\"\\r\\nContent-Type: \", content_type])\n            parts.append(b\"\\r\\n\\r\\n\")\n            self._headers = b\"\".join(parts)\n\n        return self._headers\n\n    def render_data(self) -> typing.Iterator[bytes]:\n        if isinstance(self.file, (str, bytes)):\n            yield to_bytes(self.file)\n            return\n\n        if hasattr(self, \"_data\"):\n            # Already rendered.\n            yield self._data\n            return\n\n        if self._consumed:  # pragma: nocover\n            self.file.seek(0)\n        self._consumed = True\n\n        for chunk in self.file:\n            yield to_bytes(chunk)\n\n    def render(self) -> typing.Iterator[bytes]:\n        yield self.render_headers()\n        yield from self.render_data()\n\n\nclass MultipartStream(SyncByteStream, AsyncByteStream):\n    \"\"\"\n    Request content as streaming multipart encoded form data.\n    \"\"\"\n\n    def __init__(self, data: dict, files: RequestFiles, boundary: bytes = None) -> None:\n        if boundary is None:\n            boundary = binascii.hexlify(os.urandom(16))\n\n        self.boundary = boundary\n        self.content_type = \"multipart/form-data; boundary=%s\" % boundary.decode(\n            \"ascii\"\n        )\n        self.fields = list(self._iter_fields(data, files))\n\n    def _iter_fields(\n        self, data: dict, files: RequestFiles\n    ) -> typing.Iterator[typing.Union[FileField, DataField]]:\n        for name, value in data.items():\n            if isinstance(value, list):\n                for item in value:\n                    yield DataField(name=name, value=item)\n            else:\n                yield DataField(name=name, value=value)\n\n        file_items = files.items() if isinstance(files, typing.Mapping) else files\n        for name, value in file_items:\n            yield FileField(name=name, value=value)\n\n    def iter_chunks(self) -> typing.Iterator[bytes]:\n        for field in self.fields:\n            yield b\"--%s\\r\\n\" % self.boundary\n            yield from field.render()\n            yield b\"\\r\\n\"\n        yield b\"--%s--\\r\\n\" % self.boundary\n\n    def iter_chunks_lengths(self) -> typing.Iterator[int]:\n        boundary_length = len(self.boundary)\n        # Follow closely what `.iter_chunks()` does.\n        for field in self.fields:\n            yield 2 + boundary_length + 2\n            yield field.get_length()\n            yield 2\n        yield 2 + boundary_length + 4\n\n    def get_content_length(self) -> int:\n        return sum(self.iter_chunks_lengths())\n\n    # Content stream interface.\n\n    def get_headers(self) -> typing.Dict[str, str]:\n        content_length = str(self.get_content_length())\n        content_type = self.content_type\n        return {\"Content-Length\": content_length, \"Content-Type\": content_type}\n\n    def __iter__(self) -> typing.Iterator[bytes]:\n        for chunk in self.iter_chunks():\n            yield chunk\n\n    async def __aiter__(self) -> typing.AsyncIterator[bytes]:\n        for chunk in self.iter_chunks():\n            yield chunk\n", "path": "httpx/_multipart.py"}], "after_files": [{"content": "\"\"\"\nType definitions for type checking purposes.\n\"\"\"\n\nimport ssl\nfrom http.cookiejar import CookieJar\nfrom typing import (\n    IO,\n    TYPE_CHECKING,\n    AsyncIterable,\n    Callable,\n    Dict,\n    Iterable,\n    List,\n    Mapping,\n    Optional,\n    Sequence,\n    Tuple,\n    Union,\n)\n\nif TYPE_CHECKING:  # pragma: no cover\n    from ._auth import Auth  # noqa: F401\n    from ._config import Proxy, Timeout  # noqa: F401\n    from ._models import URL, Cookies, Headers, QueryParams, Request  # noqa: F401\n\n\nPrimitiveData = Optional[Union[str, int, float, bool]]\n\nRawURL = Tuple[bytes, bytes, Optional[int], bytes]\n\nURLTypes = Union[\"URL\", str]\n\nQueryParamTypes = Union[\n    \"QueryParams\",\n    Mapping[str, Union[PrimitiveData, Sequence[PrimitiveData]]],\n    List[Tuple[str, PrimitiveData]],\n    Tuple[Tuple[str, PrimitiveData], ...],\n    str,\n    bytes,\n    None,\n]\n\nHeaderTypes = Union[\n    \"Headers\",\n    Dict[str, str],\n    Dict[bytes, bytes],\n    Sequence[Tuple[str, str]],\n    Sequence[Tuple[bytes, bytes]],\n]\n\nCookieTypes = Union[\"Cookies\", CookieJar, Dict[str, str], List[Tuple[str, str]]]\n\nCertTypes = Union[\n    # certfile\n    str,\n    # (certfile, keyfile)\n    Tuple[str, Optional[str]],\n    # (certfile, keyfile, password)\n    Tuple[str, Optional[str], Optional[str]],\n]\nVerifyTypes = Union[str, bool, ssl.SSLContext]\nTimeoutTypes = Union[\n    Optional[float],\n    Tuple[Optional[float], Optional[float], Optional[float], Optional[float]],\n    \"Timeout\",\n]\nProxiesTypes = Union[URLTypes, \"Proxy\", Dict[URLTypes, Union[None, URLTypes, \"Proxy\"]]]\n\nAuthTypes = Union[\n    Tuple[Union[str, bytes], Union[str, bytes]],\n    Callable[[\"Request\"], \"Request\"],\n    \"Auth\",\n    None,\n]\n\nRequestContent = Union[str, bytes, Iterable[bytes], AsyncIterable[bytes]]\nResponseContent = Union[str, bytes, Iterable[bytes], AsyncIterable[bytes]]\n\nRequestData = dict\n\nFileContent = Union[IO[bytes], bytes]\nFileTypes = Union[\n    # file (or text)\n    FileContent,\n    # (filename, file (or text))\n    Tuple[Optional[str], FileContent],\n    # (filename, file (or text), content_type)\n    Tuple[Optional[str], FileContent, Optional[str]],\n]\nRequestFiles = Union[Mapping[str, FileTypes], Sequence[Tuple[str, FileTypes]]]\n", "path": "httpx/_types.py"}, {"content": "import binascii\nimport io\nimport os\nimport typing\nfrom pathlib import Path\n\nfrom ._transports.base import AsyncByteStream, SyncByteStream\nfrom ._types import FileContent, FileTypes, RequestFiles\nfrom ._utils import (\n    format_form_param,\n    guess_content_type,\n    peek_filelike_length,\n    primitive_value_to_str,\n    to_bytes,\n)\n\n\nclass DataField:\n    \"\"\"\n    A single form field item, within a multipart form field.\n    \"\"\"\n\n    def __init__(\n        self, name: str, value: typing.Union[str, bytes, int, float, None]\n    ) -> None:\n        if not isinstance(name, str):\n            raise TypeError(\n                f\"Invalid type for name. Expected str, got {type(name)}: {name!r}\"\n            )\n        if value is not None and not isinstance(value, (str, bytes, int, float)):\n            raise TypeError(\n                f\"Invalid type for value. Expected primitive type, got {type(value)}: {value!r}\"\n            )\n        self.name = name\n        self.value: typing.Union[str, bytes] = (\n            value if isinstance(value, bytes) else primitive_value_to_str(value)\n        )\n\n    def render_headers(self) -> bytes:\n        if not hasattr(self, \"_headers\"):\n            name = format_form_param(\"name\", self.name)\n            self._headers = b\"\".join(\n                [b\"Content-Disposition: form-data; \", name, b\"\\r\\n\\r\\n\"]\n            )\n\n        return self._headers\n\n    def render_data(self) -> bytes:\n        if not hasattr(self, \"_data\"):\n            self._data = to_bytes(self.value)\n\n        return self._data\n\n    def get_length(self) -> int:\n        headers = self.render_headers()\n        data = self.render_data()\n        return len(headers) + len(data)\n\n    def render(self) -> typing.Iterator[bytes]:\n        yield self.render_headers()\n        yield self.render_data()\n\n\nclass FileField:\n    \"\"\"\n    A single file field item, within a multipart form field.\n    \"\"\"\n\n    def __init__(self, name: str, value: FileTypes) -> None:\n        self.name = name\n\n        fileobj: FileContent\n\n        if isinstance(value, tuple):\n            try:\n                filename, fileobj, content_type = value  # type: ignore\n            except ValueError:\n                filename, fileobj = value  # type: ignore\n                content_type = guess_content_type(filename)\n        else:\n            filename = Path(str(getattr(value, \"name\", \"upload\"))).name\n            fileobj = value\n            content_type = guess_content_type(filename)\n\n        if isinstance(fileobj, str) or isinstance(fileobj, io.StringIO):\n            raise TypeError(f\"Expected bytes or bytes-like object got: {type(fileobj)}\")\n\n        self.filename = filename\n        self.file = fileobj\n        self.content_type = content_type\n        self._consumed = False\n\n    def get_length(self) -> int:\n        headers = self.render_headers()\n\n        if isinstance(self.file, (str, bytes)):\n            return len(headers) + len(to_bytes(self.file))\n\n        # Let's do our best not to read `file` into memory.\n        file_length = peek_filelike_length(self.file)\n        if file_length is None:\n            # As a last resort, read file and cache contents for later.\n            assert not hasattr(self, \"_data\")\n            self._data = to_bytes(self.file.read())\n            file_length = len(self._data)\n\n        return len(headers) + file_length\n\n    def render_headers(self) -> bytes:\n        if not hasattr(self, \"_headers\"):\n            parts = [\n                b\"Content-Disposition: form-data; \",\n                format_form_param(\"name\", self.name),\n            ]\n            if self.filename:\n                filename = format_form_param(\"filename\", self.filename)\n                parts.extend([b\"; \", filename])\n            if self.content_type is not None:\n                content_type = self.content_type.encode()\n                parts.extend([b\"\\r\\nContent-Type: \", content_type])\n            parts.append(b\"\\r\\n\\r\\n\")\n            self._headers = b\"\".join(parts)\n\n        return self._headers\n\n    def render_data(self) -> typing.Iterator[bytes]:\n        if isinstance(self.file, (str, bytes)):\n            yield to_bytes(self.file)\n            return\n\n        if hasattr(self, \"_data\"):\n            # Already rendered.\n            yield self._data\n            return\n\n        if self._consumed:  # pragma: nocover\n            self.file.seek(0)\n        self._consumed = True\n\n        for chunk in self.file:\n            yield to_bytes(chunk)\n\n    def render(self) -> typing.Iterator[bytes]:\n        yield self.render_headers()\n        yield from self.render_data()\n\n\nclass MultipartStream(SyncByteStream, AsyncByteStream):\n    \"\"\"\n    Request content as streaming multipart encoded form data.\n    \"\"\"\n\n    def __init__(self, data: dict, files: RequestFiles, boundary: bytes = None) -> None:\n        if boundary is None:\n            boundary = binascii.hexlify(os.urandom(16))\n\n        self.boundary = boundary\n        self.content_type = \"multipart/form-data; boundary=%s\" % boundary.decode(\n            \"ascii\"\n        )\n        self.fields = list(self._iter_fields(data, files))\n\n    def _iter_fields(\n        self, data: dict, files: RequestFiles\n    ) -> typing.Iterator[typing.Union[FileField, DataField]]:\n        for name, value in data.items():\n            if isinstance(value, list):\n                for item in value:\n                    yield DataField(name=name, value=item)\n            else:\n                yield DataField(name=name, value=value)\n\n        file_items = files.items() if isinstance(files, typing.Mapping) else files\n        for name, value in file_items:\n            yield FileField(name=name, value=value)\n\n    def iter_chunks(self) -> typing.Iterator[bytes]:\n        for field in self.fields:\n            yield b\"--%s\\r\\n\" % self.boundary\n            yield from field.render()\n            yield b\"\\r\\n\"\n        yield b\"--%s--\\r\\n\" % self.boundary\n\n    def iter_chunks_lengths(self) -> typing.Iterator[int]:\n        boundary_length = len(self.boundary)\n        # Follow closely what `.iter_chunks()` does.\n        for field in self.fields:\n            yield 2 + boundary_length + 2\n            yield field.get_length()\n            yield 2\n        yield 2 + boundary_length + 4\n\n    def get_content_length(self) -> int:\n        return sum(self.iter_chunks_lengths())\n\n    # Content stream interface.\n\n    def get_headers(self) -> typing.Dict[str, str]:\n        content_length = str(self.get_content_length())\n        content_type = self.content_type\n        return {\"Content-Length\": content_length, \"Content-Type\": content_type}\n\n    def __iter__(self) -> typing.Iterator[bytes]:\n        for chunk in self.iter_chunks():\n            yield chunk\n\n    async def __aiter__(self) -> typing.AsyncIterator[bytes]:\n        for chunk in self.iter_chunks():\n            yield chunk\n", "path": "httpx/_multipart.py"}]}