(
env: Environment,
raw_body: Union[str, bytes, IO, 'MultipartEncoder', RequestDataDict],
body_read_callback: CallbackT,
offline: bool = False,
chunked: bool = False,
content_length_header_value: Optional[int] = None,
)
| 189 | |
| 190 | |
| 191 | def prepare_request_body( |
| 192 | env: Environment, |
| 193 | raw_body: Union[str, bytes, IO, 'MultipartEncoder', RequestDataDict], |
| 194 | body_read_callback: CallbackT, |
| 195 | offline: bool = False, |
| 196 | chunked: bool = False, |
| 197 | content_length_header_value: Optional[int] = None, |
| 198 | ) -> Union[bytes, IO, 'MultipartEncoder', ChunkedStream]: |
| 199 | is_file_like = hasattr(raw_body, 'read') |
| 200 | if isinstance(raw_body, (bytes, str)): |
| 201 | body = as_bytes(raw_body) |
| 202 | elif isinstance(raw_body, RequestDataDict): |
| 203 | body = as_bytes(urlencode(raw_body, doseq=True)) |
| 204 | else: |
| 205 | body = raw_body |
| 206 | |
| 207 | if offline: |
| 208 | if is_file_like: |
| 209 | return as_bytes(raw_body.read()) |
| 210 | else: |
| 211 | return body |
| 212 | |
| 213 | if is_file_like: |
| 214 | return _prepare_file_for_upload( |
| 215 | env, |
| 216 | body, |
| 217 | chunked=chunked, |
| 218 | callback=body_read_callback, |
| 219 | content_length_header_value=content_length_header_value |
| 220 | ) |
| 221 | elif chunked: |
| 222 | return ChunkedUploadStream( |
| 223 | stream=iter([body]), |
| 224 | callback=body_read_callback |
| 225 | ) |
| 226 | else: |
| 227 | return body |
| 228 | |
| 229 | |
| 230 | def get_multipart_data_and_content_type( |
no test coverage detected