An uploaded file included as part of the request data.
| 408 | |
| 409 | |
| 410 | class UploadFile: |
| 411 | class="st">""" |
| 412 | An uploaded file included as part of the request data. |
| 413 | class="st">""" |
| 414 | |
| 415 | def __init__( |
| 416 | self, |
| 417 | file: BinaryIO, |
| 418 | *, |
| 419 | size: int | None = None, |
| 420 | filename: str | None = None, |
| 421 | headers: Headers | None = None, |
| 422 | ) -> None: |
| 423 | self.filename = filename |
| 424 | self.file = file |
| 425 | self.size = size |
| 426 | self.headers = headers or Headers() |
| 427 | |
| 428 | class="cm"># Capture max size from SpooledTemporaryFile if one is provided. This slightly speeds up future checks. |
| 429 | class="cm"># Note 0 means unlimited mirroring SpooledTemporaryFile's __init__ |
| 430 | self._max_mem_size = getattr(self.file, class="st">"_max_size", 0) |
| 431 | |
| 432 | @property |
| 433 | def content_type(self) -> str | None: |
| 434 | return self.headers.get(class="st">"content-type", None) |
| 435 | |
| 436 | @property |
| 437 | def _in_memory(self) -> bool: |
| 438 | class="cm"># check for SpooledTemporaryFile._rolled |
| 439 | rolled_to_disk = getattr(self.file, class="st">"_rolled", True) |
| 440 | return not rolled_to_disk |
| 441 | |
| 442 | def _will_roll(self, size_to_add: int) -> bool: |
| 443 | class="cm"># If we're not in_memory then we will always roll |
| 444 | if not self._in_memory: |
| 445 | return True |
| 446 | |
| 447 | class="cm"># Check for SpooledTemporaryFile._max_size |
| 448 | future_size = self.file.tell() + size_to_add |
| 449 | return bool(future_size > self._max_mem_size) if self._max_mem_size else False |
| 450 | |
| 451 | async def write(self, data: bytes) -> None: |
| 452 | new_data_len = len(data) |
| 453 | if self.size is not None: |
| 454 | self.size += new_data_len |
| 455 | |
| 456 | if self._will_roll(new_data_len): |
| 457 | await run_in_threadpool(self.file.write, data) |
| 458 | else: |
| 459 | self.file.write(data) |
| 460 | |
| 461 | async def read(self, size: int = -1) -> bytes: |
| 462 | if self._in_memory: |
| 463 | return self.file.read(size) |
| 464 | return await run_in_threadpool(self.file.read, size) |
| 465 | |
| 466 | async def seek(self, offset: int) -> None: |
| 467 | if self._in_memory: |
no outgoing calls