Class MultiPartParser

src/quart/formparser.py:133–246

class MultiPartParser:
    def __init__(
        self,
        *,
        buffer_size: int = 64 * 1024,
        cls: type[MultiDict] = MultiDict,
        file_storage_cls: type[FileStorage] = FileStorage,
        max_content_length: int | None = None,
        max_form_memory_size: int | None = None,
        max_form_parts: int | None = None,
        stream_factory: StreamFactory = default_stream_factory,
    ) -> None:
        self.buffer_size = buffer_size
        self.cls = cls
        self.file_storage_cls = file_storage_cls
        self.max_content_length = max_content_length
        self.max_form_memory_size = max_form_memory_size
        self.max_form_parts = max_form_parts
        self.stream_factory = stream_factory

    def fail(self, message: str) -> NoReturn:
        raise ValueError(message)

    def get_part_charset(self, headers: Headers) -> str:
        content_type = headers.get("content-type")

        if content_type:
            parameters = parse_options_header(content_type)[1]
            ct_charset = parameters.get("charset", "").lower()

            # A safe list of encodings. Modern clients should only send ASCII or UTF-8.
            # This list will not be extended further.
            if ct_charset in {"ascii", "us-ascii", "utf-8", "iso-8859-1"}:
                return ct_charset

        return "utf-8"

    def start_file_streaming(self, event: File, total_content_length: int) -> IO[bytes]:
        content_type = event.headers.get("content-type")

        try:
            content_length = int(event.headers["content-length"])
        except (KeyError, ValueError):
            content_length = 0

        container = self.stream_factory(
            total_content_length,
            content_type,
            event.filename,
            content_length,
        )
        return container

    async def parse(
        self, body: Body, boundary: bytes, content_length: int
    ) -> tuple[MultiDict, MultiDict]:
        container: IO[bytes] | list[bytes]
        _write: Callable[[bytes], Any]