Shortcuts on this page
r m x toggle line displays
j k next/prev highlighted chunk
0 (zero) top of page
1 (one) first highlighted chunk
1from __future__ import annotations
3import codecs
4import sys
5from collections.abc import AsyncGenerator
6from dataclasses import dataclass, field
7from enum import Enum
8from tempfile import SpooledTemporaryFile
9from typing import TYPE_CHECKING
10from urllib.parse import unquote_plus
12from starlette.datastructures import FormData, Headers, UploadFile
13from starlette.exceptions import StarletteDeprecationWarning
15if sys.version_info >= (3, 13): # pragma: no cover
16 from warnings import deprecated
17else: # pragma: no cover
18 from typing_extensions import deprecated
20if TYPE_CHECKING:
21 import python_multipart as multipart
22 from python_multipart.exceptions import FormParserError
23 from python_multipart.multipart import MultipartCallbacks, QuerystringCallbacks, parse_options_header
24else:
25 try:
26 try:
27 import python_multipart as multipart
28 from python_multipart.exceptions import FormParserError
29 from python_multipart.multipart import parse_options_header
30 except ModuleNotFoundError: # pragma: no cover
31 import multipart
32 from multipart.exceptions import FormParserError
33 from multipart.multipart import parse_options_header
34 except ModuleNotFoundError: # pragma: no cover
35 multipart = None
36 parse_options_header = None
39@deprecated(
40 "`FormMessage` is deprecated and will be removed in a future version.",
41 category=StarletteDeprecationWarning,
42 stacklevel=2,
43)
44class FormMessage(Enum):
45 FIELD_START = 1
46 FIELD_NAME = 2
47 FIELD_DATA = 3
48 FIELD_END = 4
49 END = 5
52@dataclass
53class MultipartPart:
54 content_disposition: bytes | None = None
55 field_name: str = ""
56 data: bytearray = field(default_factory=bytearray)
57 file: UploadFile | None = None
58 item_headers: list[tuple[bytes, bytes]] = field(default_factory=list)
61def _user_safe_decode(src: bytes | bytearray, codec: str) -> str:
62 try:
63 return src.decode(codec)
64 except (UnicodeDecodeError, LookupError):
65 return src.decode("latin-1")
68class MultiPartException(Exception):
69 def __init__(self, message: str) -> None:
70 self.message = message
73class FormParser:
74 def __init__(
75 self,
76 headers: Headers,
77 stream: AsyncGenerator[bytes, None],
78 *,
79 max_fields: int | float = 1000,
80 max_part_size: int = 1024 * 1024, # 1MB
81 ) -> None:
82 assert multipart is not None, "The `python-multipart` library must be installed to use form parsing."
83 self.headers = headers
84 self.stream = stream
85 self.max_fields = max_fields
86 self.max_part_size = max_part_size
87 self._current_field_size = 0
88 self._current_fields = 0
89 self._field_name = bytearray()
90 self._field_value = bytearray()
91 self._items: list[tuple[str, str | UploadFile]] = []
93 def on_field_start(self) -> None:
94 self._current_field_size = 0
95 self._field_name = bytearray()
96 self._field_value = bytearray()
98 def on_field_name(self, data: bytes, start: int, end: int) -> None:
99 self._current_field_size += end - start
100 if self._current_field_size > self.max_part_size:
101 raise MultiPartException(f"Field exceeded maximum size of {int(self.max_part_size / 1024)}KB.")
102 self._field_name += data[start:end]
104 def on_field_data(self, data: bytes, start: int, end: int) -> None:
105 self._current_field_size += end - start
106 if self._current_field_size > self.max_part_size:
107 raise MultiPartException(f"Field exceeded maximum size of {int(self.max_part_size / 1024)}KB.")
108 self._field_value += data[start:end]
110 def on_field_end(self) -> None:
111 self._current_fields += 1
112 if self._current_fields > self.max_fields:
113 raise MultiPartException(f"Too many fields. Maximum number of fields is {self.max_fields}.")
114 name = unquote_plus(self._field_name.decode("latin-1"))
115 value = unquote_plus(self._field_value.decode("latin-1"))
116 self._items.append((name, value))
118 async def parse(self) -> FormData:
119 callbacks: QuerystringCallbacks = {
120 "on_field_start": self.on_field_start,
121 "on_field_name": self.on_field_name,
122 "on_field_data": self.on_field_data,
123 "on_field_end": self.on_field_end,
124 }
125 parser = multipart.QuerystringParser(callbacks)
127 async for chunk in self.stream:
128 parser.write(chunk)
129 parser.finalize()
131 return FormData(self._items)
134class MultiPartParser:
135 spool_max_size = 1024 * 1024 # 1MB
136 """The maximum size of the spooled temporary file used to store file data."""
137 max_part_size = 1024 * 1024 # 1MB
138 """The maximum size of a part in the multipart request."""
140 def __init__(
141 self,
142 headers: Headers,
143 stream: AsyncGenerator[bytes, None],
144 *,
145 max_files: int | float = 1000,
146 max_fields: int | float = 1000,
147 max_part_size: int = 1024 * 1024, # 1MB
148 ) -> None:
149 assert multipart is not None, "The `python-multipart` library must be installed to use form parsing."
150 self.headers = headers
151 self.stream = stream
152 self.max_files = max_files
153 self.max_fields = max_fields
154 self.items: list[tuple[str, str | UploadFile]] = []
155 self._current_files = 0
156 self._current_fields = 0
157 self._current_partial_header_name: bytes = b""
158 self._current_partial_header_value: bytes = b""
159 self._current_part = MultipartPart()
160 self._charset = ""
161 self._file_parts_to_write: list[tuple[MultipartPart, bytes]] = []
162 self._file_parts_to_finish: list[MultipartPart] = []
163 self._files_to_close_on_error: list[SpooledTemporaryFile[bytes]] = []
164 self.max_part_size = max_part_size
166 def on_part_begin(self) -> None:
167 self._current_part = MultipartPart()
169 def on_part_data(self, data: bytes, start: int, end: int) -> None:
170 message_bytes = data[start:end]
171 if self._current_part.file is None:
172 if len(self._current_part.data) + len(message_bytes) > self.max_part_size:
173 raise MultiPartException(f"Part exceeded maximum size of {int(self.max_part_size / 1024)}KB.")
174 self._current_part.data.extend(message_bytes)
175 else:
176 self._file_parts_to_write.append((self._current_part, message_bytes))
178 def on_part_end(self) -> None:
179 if self._current_part.file is None:
180 self.items.append(
181 (
182 self._current_part.field_name,
183 _user_safe_decode(self._current_part.data, self._charset),
184 )
185 )
186 else:
187 self._file_parts_to_finish.append(self._current_part)
188 # The file can be added to the items right now even though it's not
189 # finished yet, because it will be finished in the `parse()` method, before
190 # self.items is used in the return value.
191 self.items.append((self._current_part.field_name, self._current_part.file))
193 def on_header_field(self, data: bytes, start: int, end: int) -> None:
194 self._current_partial_header_name += data[start:end]
196 def on_header_value(self, data: bytes, start: int, end: int) -> None:
197 self._current_partial_header_value += data[start:end]
199 def on_header_end(self) -> None:
200 field = self._current_partial_header_name.lower()
201 if field == b"content-disposition":
202 self._current_part.content_disposition = self._current_partial_header_value
203 self._current_part.item_headers.append((field, self._current_partial_header_value))
204 self._current_partial_header_name = b""
205 self._current_partial_header_value = b""
207 def on_headers_finished(self) -> None:
208 disposition, options = parse_options_header(self._current_part.content_disposition)
209 try:
210 self._current_part.field_name = _user_safe_decode(options[b"name"], self._charset)
211 except KeyError:
212 raise MultiPartException('The Content-Disposition header field "name" must be provided.')
213 if b"filename" in options:
214 self._current_files += 1
215 if self._current_files > self.max_files:
216 raise MultiPartException(f"Too many files. Maximum number of files is {self.max_files}.")
217 filename = _user_safe_decode(options[b"filename"], self._charset)
218 tempfile = SpooledTemporaryFile(max_size=self.spool_max_size)
219 self._files_to_close_on_error.append(tempfile)
220 self._current_part.file = UploadFile(
221 file=tempfile, # type: ignore[arg-type]
222 size=0,
223 filename=filename,
224 headers=Headers(raw=self._current_part.item_headers),
225 )
226 else:
227 self._current_fields += 1
228 if self._current_fields > self.max_fields:
229 raise MultiPartException(f"Too many fields. Maximum number of fields is {self.max_fields}.")
230 self._current_part.file = None
232 def on_end(self) -> None:
233 pass
235 async def parse(self) -> FormData:
236 # Parse the Content-Type header to get the multipart boundary.
237 _, params = parse_options_header(self.headers["Content-Type"])
238 charset = params.get(b"charset", "utf-8")
239 if isinstance(charset, bytes):
240 charset = charset.decode("latin-1")
241 try:
242 self._charset = codecs.lookup(charset).name
243 except LookupError:
244 self._charset = "latin-1"
245 try:
246 boundary = params[b"boundary"]
247 except KeyError:
248 raise MultiPartException("Missing boundary in multipart.")
250 # Callbacks dictionary.
251 callbacks: MultipartCallbacks = {
252 "on_part_begin": self.on_part_begin,
253 "on_part_data": self.on_part_data,
254 "on_part_end": self.on_part_end,
255 "on_header_field": self.on_header_field,
256 "on_header_value": self.on_header_value,
257 "on_header_end": self.on_header_end,
258 "on_headers_finished": self.on_headers_finished,
259 "on_end": self.on_end,
260 }
262 try:
263 parser = multipart.MultipartParser(boundary, callbacks)
264 # Feed the parser with data from the request.
265 async for chunk in self.stream:
266 parser.write(chunk)
267 # Write file data, it needs to use await with the UploadFile methods
268 # that call the corresponding file methods *in a threadpool*,
269 # otherwise, if they were called directly in the callback methods above
270 # (regular, non-async functions), that would block the event loop in
271 # the main thread.
272 for part, data in self._file_parts_to_write:
273 assert part.file # for type checkers
274 await part.file.write(data)
275 for part in self._file_parts_to_finish:
276 assert part.file # for type checkers
277 await part.file.seek(0)
278 self._file_parts_to_write.clear()
279 self._file_parts_to_finish.clear()
280 parser.finalize()
281 except BaseException as exc:
282 # Close all the files if parsing or reading the request stream fails.
283 for file in self._files_to_close_on_error:
284 file.close()
285 if isinstance(exc, FormParserError):
286 raise MultiPartException("Invalid multipart data.") from exc
287 raise
289 return FormData(self.items)