(
content: str | bytes | Iterable[bytes] | AsyncIterable[bytes],
)
| 105 | |
| 106 | |
| 107 | def encode_content( |
| 108 | content: str | bytes | Iterable[bytes] | AsyncIterable[bytes], |
| 109 | ) -> tuple[dict[str, str], SyncByteStream | AsyncByteStream]: |
| 110 | if isinstance(content, (bytes, str)): |
| 111 | body = content.encode("utf-8") if isinstance(content, str) else content |
| 112 | content_length = len(body) |
| 113 | headers = {"Content-Length": str(content_length)} if body else {} |
| 114 | return headers, ByteStream(body) |
| 115 | |
| 116 | elif isinstance(content, Iterable) and not isinstance(content, dict): |
| 117 | # `not isinstance(content, dict)` is a bit oddly specific, but it |
| 118 | # catches a case that's easy for users to make in error, and would |
| 119 | # otherwise pass through here, like any other bytes-iterable, |
| 120 | # because `dict` happens to be iterable. See issue #2491. |
| 121 | content_length_or_none = peek_filelike_length(content) |
| 122 | |
| 123 | if content_length_or_none is None: |
| 124 | headers = {"Transfer-Encoding": "chunked"} |
| 125 | else: |
| 126 | headers = {"Content-Length": str(content_length_or_none)} |
| 127 | return headers, IteratorByteStream(content) # type: ignore |
| 128 | |
| 129 | elif isinstance(content, AsyncIterable): |
| 130 | headers = {"Transfer-Encoding": "chunked"} |
| 131 | return headers, AsyncIteratorByteStream(content) |
| 132 | |
| 133 | raise TypeError(f"Unexpected type for 'content', {type(content)!r}") |
| 134 | |
| 135 | |
| 136 | def encode_urlencoded_data( |
no test coverage detected