| 58 | |
| 59 | |
| 60 | class FormParser: |
| 61 | def __init__( |
| 62 | self, |
| 63 | headers: Headers, |
| 64 | stream: AsyncGenerator[bytes, None], |
| 65 | *, |
| 66 | max_fields: int | float = 1000, |
| 67 | max_part_size: int = 1024 * 1024, # 1MB |
| 68 | ) -> None: |
| 69 | assert multipart is not None, "The `python-multipart` library must be installed to use form parsing." |
| 70 | self.headers = headers |
| 71 | self.stream = stream |
| 72 | self.max_fields = max_fields |
| 73 | self.max_part_size = max_part_size |
| 74 | self.messages: list[tuple[FormMessage, bytes]] = [] |
| 75 | self._current_field_size = 0 |
| 76 | self._current_fields = 0 |
| 77 | |
| 78 | def on_field_start(self) -> None: |
| 79 | self._current_field_size = 0 |
| 80 | message = (FormMessage.FIELD_START, b"") |
| 81 | self.messages.append(message) |
| 82 | |
| 83 | def on_field_name(self, data: bytes, start: int, end: int) -> None: |
| 84 | self._current_field_size += end - start |
| 85 | if self._current_field_size > self.max_part_size: |
| 86 | raise MultiPartException(f"Field exceeded maximum size of {int(self.max_part_size / 1024)}KB.") |
| 87 | message = (FormMessage.FIELD_NAME, data[start:end]) |
| 88 | self.messages.append(message) |
| 89 | |
| 90 | def on_field_data(self, data: bytes, start: int, end: int) -> None: |
| 91 | self._current_field_size += end - start |
| 92 | if self._current_field_size > self.max_part_size: |
| 93 | raise MultiPartException(f"Field exceeded maximum size of {int(self.max_part_size / 1024)}KB.") |
| 94 | message = (FormMessage.FIELD_DATA, data[start:end]) |
| 95 | self.messages.append(message) |
| 96 | |
| 97 | def on_field_end(self) -> None: |
| 98 | self._current_fields += 1 |
| 99 | if self._current_fields > self.max_fields: |
| 100 | raise MultiPartException(f"Too many fields. Maximum number of fields is {self.max_fields}.") |
| 101 | message = (FormMessage.FIELD_END, b"") |
| 102 | self.messages.append(message) |
| 103 | |
| 104 | def on_end(self) -> None: |
| 105 | message = (FormMessage.END, b"") |
| 106 | self.messages.append(message) |
| 107 | |
| 108 | async def parse(self) -> FormData: |
| 109 | # Callbacks dictionary. |
| 110 | callbacks: QuerystringCallbacks = { |
| 111 | "on_field_start": self.on_field_start, |
| 112 | "on_field_name": self.on_field_name, |
| 113 | "on_field_data": self.on_field_data, |
| 114 | "on_field_end": self.on_field_end, |
| 115 | "on_end": self.on_end, |
| 116 | } |
| 117 |
no outgoing calls