diff --git a/CHANGELOG.md b/CHANGELOG.md index aeb17fe..f074cbe 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -3,6 +3,7 @@ ## Unreleased * Speed up querystring callback dispatch [#316](https://github.com/Kludex/python-multipart/pull/316). +* Speed up `MultipartParser` callback dispatch [#337](https://github.com/Kludex/python-multipart/pull/337). ## 0.0.32 (2026-06-04) diff --git a/python_multipart/multipart.py b/python_multipart/multipart.py index cfc3220..336b94e 100644 --- a/python_multipart/multipart.py +++ b/python_multipart/multipart.py @@ -1140,6 +1140,39 @@ def _internal_write(self, data: bytes, length: int) -> int: current_header_count = self._current_header_count current_header_size = self._current_header_size + # Hoist callback dispatch out of the per-byte loop (same shape as the + # QuerystringParser optimization in #316): one dict lookup per write + # instead of a `"on_" + name` concatenation, dict lookup and `cast` + # per emitted event. + callbacks = cast("MultipartCallbacks", self.callbacks) + on_part_begin = callbacks.get("on_part_begin") + on_part_data = callbacks.get("on_part_data") + on_part_end = callbacks.get("on_part_end") + on_header_begin = callbacks.get("on_header_begin") + on_header_field = callbacks.get("on_header_field") + on_header_value = callbacks.get("on_header_value") + on_header_end = callbacks.get("on_header_end") + on_headers_finished = callbacks.get("on_headers_finished") + on_end = callbacks.get("on_end") + if on_part_begin is None: + on_part_begin = _noop_event + if on_part_data is None: + on_part_data = _noop_data + if on_part_end is None: + on_part_end = _noop_event + if on_header_begin is None: + on_header_begin = _noop_event + if on_header_field is None: + on_header_field = _noop_data + if on_header_value is None: + on_header_value = _noop_data + if on_header_end is None: + on_header_end = _noop_event + if on_headers_finished is None: + on_headers_finished = _noop_event + if on_end is None: + on_end = _noop_event + # Our index defaults to 0. i = 0 @@ -1162,7 +1195,7 @@ def delete_mark(name: str, reset: bool = False) -> None: # end of the buffer, and reset the mark, instead of deleting it. This # is used at the end of the function to call our callbacks with any # remaining data in this chunk. - def data_callback(name: CallbackName, end_i: int, remaining: bool = False) -> None: + def data_callback(name: CallbackName, end_i: int, func: Callable[..., Any], remaining: bool = False) -> None: marked_index = self.marks.get(name) if marked_index is None: return @@ -1174,7 +1207,7 @@ def data_callback(name: CallbackName, end_i: int, remaining: bool = False) -> No pass elif marked_index >= 0: # We are emitting data from the local buffer. - self.callback(name, data, marked_index, end_i) + func(data, marked_index, end_i) else: # Some of the data comes from a partial boundary match. # and requires look-behind. @@ -1182,18 +1215,18 @@ def data_callback(name: CallbackName, end_i: int, remaining: bool = False) -> No # the state when we entered the loop. lookbehind_len = -marked_index if lookbehind_len <= boundary_length: - self.callback(name, boundary, 0, lookbehind_len) + func(boundary, 0, lookbehind_len) elif self.flags & FLAG_PART_BOUNDARY: lookback = boundary + b"\r\n" - self.callback(name, lookback, 0, lookbehind_len) + func(lookback, 0, lookbehind_len) elif self.flags & FLAG_LAST_BOUNDARY: lookback = boundary + b"--\r\n" - self.callback(name, lookback, 0, lookbehind_len) + func(lookback, 0, lookbehind_len) else: # pragma: no cover (error case) self.logger.warning("Look-back buffer error") if end_i > 0: - self.callback(name, data, 0, end_i) + func(data, 0, end_i) # If we're getting remaining data, we have got all the data we # can be certain is not a boundary, leaving only a partial boundary match. if remaining: @@ -1253,7 +1286,7 @@ def data_callback(name: CallbackName, end_i: int, remaining: bool = False) -> No index = 0 # Callback for the start of a part. - self.callback("part_begin") + on_part_begin() current_header_count = 0 current_header_size = 0 @@ -1289,7 +1322,7 @@ def data_callback(name: CallbackName, end_i: int, remaining: bool = False) -> No # to stop parsing headers in the MultipartState.HEADER_FIELD state, # below. if c != CR: - self.callback("header_begin") + on_header_begin() # Move to parsing header fields. state = HEADER_FIELD @@ -1335,7 +1368,7 @@ def data_callback(name: CallbackName, end_i: int, remaining: bool = False) -> No # Call our callback with the header field. i = colon - data_callback("header_field", i) + data_callback("header_field", i, on_header_field) # Move to parsing the header value. state = HEADER_VALUE_START @@ -1362,8 +1395,8 @@ def data_callback(name: CallbackName, end_i: int, remaining: bool = False) -> No advance_header_size(end - i) if cr != -1: i = cr - data_callback("header_value", i) - self.callback("header_end") + data_callback("header_value", i, on_header_value) + on_header_end() current_header_size = 0 state = HEADER_VALUE_ALMOST_DONE else: @@ -1390,7 +1423,7 @@ def data_callback(name: CallbackName, end_i: int, remaining: bool = False) -> No self.logger.warning(msg) raise MultipartParseError(msg, offset=i) - self.callback("headers_finished") + on_headers_finished() state = PART_DATA_START elif state == PART_DATA_START: @@ -1477,11 +1510,11 @@ def data_callback(name: CallbackName, end_i: int, remaining: bool = False) -> No flags &= ~FLAG_PART_BOUNDARY # We have identified a boundary, callback for any data before it. - data_callback("part_data", i - index) + data_callback("part_data", i - index, on_part_data) # Callback indicating that we've reached the end of # a part, and are starting a new one. - self.callback("part_end") - self.callback("part_begin") + on_part_end() + on_part_begin() current_header_count = 0 current_header_size = 0 @@ -1502,11 +1535,11 @@ def data_callback(name: CallbackName, end_i: int, remaining: bool = False) -> No # We need a second hyphen here. if c == HYPHEN: # We have identified a boundary, callback for any data before it. - data_callback("part_data", i - index) + data_callback("part_data", i - index, on_part_data) # Callback to end the current part, and then the # message. - self.callback("part_end") - self.callback("end") + on_part_end() + on_end() state = END else: # No match, so reset index. @@ -1531,7 +1564,7 @@ def data_callback(name: CallbackName, end_i: int, remaining: bool = False) -> No self.logger.warning(msg) raise MultipartParseError(msg, offset=i) index += 1 - self.callback("end") + on_end() state = END elif state == END: @@ -1557,9 +1590,9 @@ def data_callback(name: CallbackName, end_i: int, remaining: bool = False) -> No # that we haven't yet reached the end of this 'thing'. So, by setting # the mark to 0, we cause any data callbacks that take place in future # calls to this function to start from the beginning of that buffer. - data_callback("header_field", length, True) - data_callback("header_value", length, True) - data_callback("part_data", length - index, True) + data_callback("header_field", length, on_header_field, True) + data_callback("header_value", length, on_header_value, True) + data_callback("part_data", length - index, on_part_data, True) # Save values to locals. self.state = state diff --git a/tests/test_multipart.py b/tests/test_multipart.py index de220eb..36b142a 100644 --- a/tests/test_multipart.py +++ b/tests/test_multipart.py @@ -644,6 +644,101 @@ def test_invalid_max_size(self) -> None: q = OctetStreamParser(max_size="foo") # type: ignore[arg-type] +class TestMultipartParser(unittest.TestCase): + BOUNDARY = b"boundary" + BODY = ( + b"--boundary\r\n" + b'Content-Disposition: form-data; name="a"\r\n' + b"\r\n" + b"foo\r\n" + b"--boundary\r\n" + b'Content-Disposition: form-data; name="b"; filename="b.txt"\r\n' + b"Content-Type: text/plain\r\n" + b"\r\n" + b"barbaz\r\n" + b"--boundary--\r\n" + ) + + def parse(self, callbacks: Any = None, chunk_size: int | None = None) -> list[tuple[Any, ...]]: + events: list[tuple[Any, ...]] = [] + + def make_event(name: str) -> Any: + def on_event() -> None: + events.append((name,)) + + return on_event + + def make_data(name: str) -> Any: + def on_data(data: bytes, start: int, end: int) -> None: + events.append((name, data[start:end])) + + return on_data + + if callbacks is None: + callbacks = { + "on_part_begin": make_event("part_begin"), + "on_part_data": make_data("part_data"), + "on_part_end": make_event("part_end"), + "on_header_begin": make_event("header_begin"), + "on_header_field": make_data("header_field"), + "on_header_value": make_data("header_value"), + "on_header_end": make_event("header_end"), + "on_headers_finished": make_event("headers_finished"), + "on_end": make_event("end"), + } + parser = MultipartParser(self.BOUNDARY, callbacks) + if chunk_size is None: + self.assertEqual(parser.write(self.BODY), len(self.BODY)) + else: + for i in range(0, len(self.BODY), chunk_size): + parser.write(self.BODY[i : i + chunk_size]) + parser.finalize() + return events + + def test_no_callbacks(self) -> None: + parser = MultipartParser(self.BOUNDARY) + + self.assertEqual(parser.write(self.BODY), len(self.BODY)) + parser.finalize() + + def test_none_callbacks(self) -> None: + callbacks: Any = { + "on_part_begin": None, + "on_part_data": None, + "on_part_end": None, + "on_header_begin": None, + "on_header_field": None, + "on_header_value": None, + "on_header_end": None, + "on_headers_finished": None, + "on_end": None, + } + parser = MultipartParser(self.BOUNDARY, callbacks) + + self.assertEqual(parser.write(self.BODY), len(self.BODY)) + parser.finalize() + + def test_events_match_across_chunkings(self) -> None: + # Chunk boundaries legitimately split data callbacks, so compare the + # joined payload per channel plus the notification-event sequence. + def summarize(events: list[tuple[Any, ...]]) -> tuple[Any, ...]: + joined: dict[str, list[bytes]] = {} + notifications: list[str] = [] + for event in events: + if len(event) == 2: + joined.setdefault(event[0], []).append(event[1]) + else: + notifications.append(event[0]) + return ({name: b"".join(payloads) for name, payloads in joined.items()}, notifications) + + whole = self.parse() + self.assertIn(("part_data", b"foo"), whole) + self.assertIn(("part_data", b"barbaz"), whole) + expected = summarize(whole) + for chunk_size in (1, 3, 7, 64): + self.assertEqual(summarize(self.parse(chunk_size=chunk_size)), expected) + + class TestBase64Decoder(unittest.TestCase): # Note: base64('foobar') == 'Zm9vYmFy' def setUp(self) -> None: