1 from _typeshed.wsgi import WSGIEnvironment
2 from typing import IO, Any, Callable, Generator, Iterable, Mapping, NoReturn, Optional, Protocol, Text, Tuple, TypeVar
4 from .datastructures import Headers
7 _ParseFunc = Callable[[IO[bytes], str, Optional[int], Mapping[str, str]], Tuple[IO[bytes], _Dict, _Dict]]
9 _F = TypeVar("_F", bound=Callable[..., Any])
11 class _StreamFactory(Protocol):
13 self, total_content_length: int | None, filename: str, content_type: str, content_length: int | None = ...
16 def default_stream_factory(
17 total_content_length: int | None, filename: str, content_type: str, content_length: int | None = ...
20 environ: WSGIEnvironment,
21 stream_factory: _StreamFactory | None = ...,
24 max_form_memory_size: int | None = ...,
25 max_content_length: int | None = ...,
26 cls: Callable[[], _Dict] | None = ...,
28 ) -> tuple[IO[bytes], _Dict, _Dict]: ...
29 def exhaust_stream(f: _F) -> _F: ...
31 class FormDataParser(object):
32 stream_factory: _StreamFactory
35 max_form_memory_size: int | None
36 max_content_length: int | None
37 cls: Callable[[], _Dict]
41 stream_factory: _StreamFactory | None = ...,
44 max_form_memory_size: int | None = ...,
45 max_content_length: int | None = ...,
46 cls: Callable[[], _Dict] | None = ...,
49 def get_parse_func(self, mimetype: str, options: Any) -> _ParseFunc | None: ...
50 def parse_from_environ(self, environ: WSGIEnvironment) -> tuple[IO[bytes], _Dict, _Dict]: ...
52 self, stream: IO[bytes], mimetype: Text, content_length: int | None, options: Mapping[str, str] | None = ...
53 ) -> tuple[IO[bytes], _Dict, _Dict]: ...
54 parse_functions: dict[Text, _ParseFunc]
56 def is_valid_multipart_boundary(boundary: str) -> bool: ...
57 def parse_multipart_headers(iterable: Iterable[Text | bytes]) -> Headers: ...
59 class MultiPartParser(object):
62 max_form_memory_size: int | None
63 stream_factory: _StreamFactory
64 cls: Callable[[], _Dict]
68 stream_factory: _StreamFactory | None = ...,
71 max_form_memory_size: int | None = ...,
72 cls: Callable[[], _Dict] | None = ...,
73 buffer_size: int = ...,
75 def fail(self, message: Text) -> NoReturn: ...
76 def get_part_encoding(self, headers: Mapping[str, str]) -> str | None: ...
77 def get_part_charset(self, headers: Mapping[str, str]) -> Text: ...
78 def start_file_streaming(
79 self, filename: Text | bytes, headers: Mapping[str, str], total_content_length: int | None
80 ) -> tuple[Text, IO[bytes]]: ...
81 def in_memory_threshold_reached(self, bytes: Any) -> NoReturn: ...
82 def validate_boundary(self, boundary: str | None) -> None: ...
84 self, file: Any, boundary: bytes, content_length: int, cap_at_buffer: bool = ...
85 ) -> Generator[tuple[str, Any], None, None]: ...
86 def parse_parts(self, file: Any, boundary: bytes, content_length: int) -> Generator[tuple[str, Any], None, None]: ...
87 def parse(self, file: Any, boundary: bytes, content_length: int) -> tuple[_Dict, _Dict]: ...