Decodes a multipart message as bytes into Python events. The part data is returned as available to allow the caller to save the data from memory to disk, if desired. .. versionchanged:: 3.1.4 Handle chunks that split a``\r\n`` sequence.
| 75 | |
| 76 | |
| 77 | class MultipartDecoder: |
| 78 | """Decodes a multipart message as bytes into Python events. |
| 79 | |
| 80 | The part data is returned as available to allow the caller to save |
| 81 | the data from memory to disk, if desired. |
| 82 | |
| 83 | .. versionchanged:: 3.1.4 |
| 84 | Handle chunks that split a``\r\n`` sequence. |
| 85 | """ |
| 86 | |
| 87 | def __init__( |
| 88 | self, |
| 89 | boundary: bytes, |
| 90 | max_form_memory_size: int | None = None, |
| 91 | *, |
| 92 | max_parts: int | None = None, |
| 93 | ) -> None: |
| 94 | self.buffer = bytearray() |
| 95 | self.complete = False |
| 96 | self.max_form_memory_size = max_form_memory_size |
| 97 | self.max_parts = max_parts |
| 98 | self.state = State.PREAMBLE |
| 99 | self.boundary = boundary |
| 100 | |
| 101 | # Note in the below \h i.e. horizontal whitespace is used |
| 102 | # as [^\S\n\r] as \h isn't supported in python. |
| 103 | |
| 104 | # The preamble must end with a boundary where the boundary is |
| 105 | # prefixed by a line break, RFC2046. Except that many |
| 106 | # implementations including Werkzeug's tests omit the line |
| 107 | # break prefix. In addition the first boundary could be the |
| 108 | # epilogue boundary (for empty form-data) hence the matching |
| 109 | # group to understand if it is an epilogue boundary. |
| 110 | self.preamble_re = re.compile( |
| 111 | rb"%s?--%s(--[^\S\n\r]*%s?|[^\S\n\r]*%s)" |
| 112 | % (LINE_BREAK, re.escape(boundary), LINE_BREAK, LINE_BREAK), |
| 113 | re.MULTILINE, |
| 114 | ) |
| 115 | # A boundary must include a line break prefix and suffix, and |
| 116 | # may include trailing whitespace. In addition the boundary |
| 117 | # could be the epilogue boundary hence the matching group to |
| 118 | # understand if it is an epilogue boundary. |
| 119 | self.boundary_re = re.compile( |
| 120 | rb"%s--%s(--[^\S\n\r]*%s?|[^\S\n\r]*%s)" |
| 121 | % (LINE_BREAK, re.escape(boundary), LINE_BREAK, LINE_BREAK), |
| 122 | re.MULTILINE, |
| 123 | ) |
| 124 | self._search_position = 0 |
| 125 | self._parts_decoded = 0 |
| 126 | |
| 127 | def receive_data(self, data: bytes | None) -> None: |
| 128 | if data is None: |
| 129 | self.complete = True |
| 130 | elif ( |
| 131 | self.max_form_memory_size is not None |
| 132 | and len(self.buffer) + len(data) > self.max_form_memory_size |
| 133 | ): |
| 134 | # Ensure that data within single event does not exceed limit. |
no outgoing calls