Adds a new body part to multipart writer.
(self, payload: Payload)
| 985 | return self.append_payload(payload) |
| 986 | |
| 987 | def append_payload(self, payload: Payload) -> Payload: |
| 988 | """Adds a new body part to multipart writer.""" |
| 989 | encoding: str | None = None |
| 990 | te_encoding: str | None = None |
| 991 | if self._is_form_data: |
| 992 | # https://datatracker.ietf.org/doc/html/rfc7578#section-4.7 |
| 993 | # https://datatracker.ietf.org/doc/html/rfc7578#section-4.8 |
| 994 | assert ( |
| 995 | not {CONTENT_ENCODING, CONTENT_LENGTH, CONTENT_TRANSFER_ENCODING} |
| 996 | & payload.headers.keys() |
| 997 | ) |
| 998 | # Set default Content-Disposition in case user doesn't create one |
| 999 | if CONTENT_DISPOSITION not in payload.headers: |
| 1000 | name = f"section-{len(self._parts)}" |
| 1001 | payload.set_content_disposition("form-data", name=name) |
| 1002 | else: |
| 1003 | # compression |
| 1004 | encoding = payload.headers.get(CONTENT_ENCODING, "").lower() |
| 1005 | if encoding and encoding not in ("deflate", "gzip", "identity"): |
| 1006 | raise RuntimeError(f"unknown content encoding: {encoding}") |
| 1007 | if encoding == "identity": |
| 1008 | encoding = None |
| 1009 | |
| 1010 | # te encoding |
| 1011 | te_encoding = payload.headers.get(CONTENT_TRANSFER_ENCODING, "").lower() |
| 1012 | if te_encoding not in ("", "base64", "quoted-printable", "binary"): |
| 1013 | raise RuntimeError(f"unknown content transfer encoding: {te_encoding}") |
| 1014 | if te_encoding == "binary": |
| 1015 | te_encoding = None |
| 1016 | |
| 1017 | # size |
| 1018 | size = payload.size |
| 1019 | if size is not None and not (encoding or te_encoding): |
| 1020 | payload.headers[CONTENT_LENGTH] = str(size) |
| 1021 | |
| 1022 | self._parts.append((payload, encoding, te_encoding)) # type: ignore[arg-type] |
| 1023 | return payload |
| 1024 | |
| 1025 | def append_json( |
| 1026 | self, obj: Any, headers: Mapping[str, str] | None = None |