A thin wrapper over incoming files.
| 10 | |
| 11 | |
| 12 | class FileStorage(WerkzeugFileStorage): |
| 13 | """A thin wrapper over incoming files.""" |
| 14 | |
| 15 | def __init__( |
| 16 | self, |
| 17 | stream: IO[bytes] | None = None, |
| 18 | filename: str | None = None, |
| 19 | name: str | None = None, |
| 20 | content_type: str | None = None, |
| 21 | content_length: int | None = None, |
| 22 | headers: Headers | None = None, |
| 23 | ) -> None: |
| 24 | super().__init__(stream, filename, name, content_type, content_length, headers) |
| 25 | |
| 26 | async def save(self, destination: PathLike, buffer_size: int = 16384) -> None: # type: ignore |
| 27 | """Save the file to the destination. |
| 28 | |
| 29 | Arguments: |
| 30 | destination: A filename (str) or file object to write to. |
| 31 | buffer_size: Buffer size to keep in memory. |
| 32 | """ |
| 33 | async with async_open(destination, "wb") as file_: |
| 34 | data = self.stream.read(buffer_size) |
| 35 | while data != b"": |
| 36 | await file_.write(data) |
| 37 | data = self.stream.read(buffer_size) |
| 38 | |
| 39 | async def load(self, source: PathLike, buffer_size: int = 16384) -> None: |
| 40 | path = Path(source) |
| 41 | self.filename = path.name |
| 42 | async with async_open(path, "rb") as file_: |
| 43 | data = await file_.read(buffer_size) |
| 44 | while data != b"": |
| 45 | self.stream.write(data) |
| 46 | data = await file_.read(buffer_size) |
no outgoing calls
searching dependent graphs…