(self, data: bytes, start: int, end: int)
| 179 | self._current_part = MultipartPart() |
| 180 | |
| 181 | def on_part_data(self, data: bytes, start: int, end: int) -> None: |
| 182 | message_bytes = data[start:end] |
| 183 | if self._current_part.file is None: |
| 184 | if len(self._current_part.data) + len(message_bytes) > self.max_part_size: |
| 185 | raise MultiPartException(fclass="st">"Part exceeded maximum size of {int(self.max_part_size / 1024)}KB.") |
| 186 | self._current_part.data.extend(message_bytes) |
| 187 | else: |
| 188 | self._file_parts_to_write.append((self._current_part, message_bytes)) |
| 189 | |
| 190 | def on_part_end(self) -> None: |
| 191 | if self._current_part.file is None: |
nothing calls this directly
no test coverage detected