From 1436cfdef94802a2d57588f277624a423b27056d Mon Sep 17 00:00:00 2001 From: Sebastian Rittau Date: Thu, 4 Jul 2019 21:58:51 +0200 Subject: [PATCH] Add types to FileStorage (#3067) --- third_party/2and3/werkzeug/datastructures.pyi | 46 ++++--- third_party/2and3/werkzeug/formparser.pyi | 116 ++++++++++++------ 2 files changed, 107 insertions(+), 55 deletions(-) diff --git a/third_party/2and3/werkzeug/datastructures.pyi b/third_party/2and3/werkzeug/datastructures.pyi index f10665f98..7ee49dda7 100644 --- a/third_party/2and3/werkzeug/datastructures.pyi +++ b/third_party/2and3/werkzeug/datastructures.pyi @@ -1,5 +1,5 @@ import collections -from typing import Any, Optional, Mapping, Dict, TypeVar, Callable, Union, overload, Text +from typing import Any, Optional, Mapping, Dict, TypeVar, Callable, Union, overload, Text, Protocol, Iterator, IO from collections import Container, Iterable, MutableSet _K = TypeVar("_K") @@ -402,24 +402,34 @@ class WWWAuthenticate(UpdateDictMixin, dict): qop: Any stale: Any -class FileStorage: - name: Any - stream: Any - filename: Any - headers: Any - def __init__(self, stream: Optional[Any] = ..., filename: Optional[Any] = ..., name: Optional[Any] = ..., - content_type: Optional[Any] = ..., content_length: Optional[Any] = ..., headers: Optional[Any] = ...): ... +class _Writer(Protocol): + def write(self, data: bytes) -> Any: ... + +class FileStorage(object): + name: Optional[Text] + stream: IO[bytes] + filename: Optional[Text] + headers: Headers + def __init__( + self, + stream: Optional[IO[bytes]] = ..., + filename: Union[None, Text, bytes] = ..., + name: Optional[Text] = ..., + content_type: Optional[Text] = ..., + content_length: Optional[int] = ..., + headers: Optional[Headers] = ..., + ): ... @property - def content_type(self): ... + def content_type(self) -> Optional[Text]: ... @property - def content_length(self): ... + def content_length(self) -> int: ... @property - def mimetype(self): ... + def mimetype(self) -> str: ... @property - def mimetype_params(self): ... - def save(self, dst, buffer_size: int = ...): ... - def close(self): ... - def __nonzero__(self): ... - __bool__: Any - def __getattr__(self, name): ... - def __iter__(self): ... + def mimetype_params(self) -> Dict[str, str]: ... + def save(self, dst: Union[Text, _Writer], buffer_size: int = ...): ... + def close(self) -> None: ... + def __nonzero__(self) -> bool: ... + def __bool__(self) -> bool: ... + def __getattr__(self, name: Text) -> Any: ... + def __iter__(self) -> Iterator[bytes]: ... diff --git a/third_party/2and3/werkzeug/formparser.pyi b/third_party/2and3/werkzeug/formparser.pyi index d0e6a7391..16c91c350 100644 --- a/third_party/2and3/werkzeug/formparser.pyi +++ b/third_party/2and3/werkzeug/formparser.pyi @@ -1,45 +1,87 @@ -from typing import Any, Optional, Text +from typing import Any, Optional, Text, Tuple, Callable, Iterable, TypeVar, NoReturn, Protocol, IO, Generator, Dict, Mapping, Union +from wsgiref.types import WSGIEnvironment -def default_stream_factory(total_content_length, filename, content_type, content_length: Optional[Any] = ...): ... -def parse_form_data(environ, stream_factory: Optional[Any] = ..., charset: Text = ..., errors: Text = ..., - max_form_memory_size: Optional[Any] = ..., max_content_length: Optional[Any] = ..., - cls: Optional[Any] = ..., silent: bool = ...): ... -def exhaust_stream(f): ... +from .datastructures import Headers -class FormDataParser: - stream_factory: Any +_Dict = Any +_ParseFunc = Callable[[IO[bytes], str, Optional[int], Mapping[str, str]], Tuple[IO[bytes], _Dict, _Dict]] + +_F = TypeVar("_F", bound=Callable[..., Any]) + +class _StreamFactory(Protocol): + def __call__( + self, total_content_length: Optional[int], filename: str, content_type: str, content_length: Optional[int] = ..., + ) -> IO[bytes]: ... + +def default_stream_factory( + total_content_length: Optional[int], filename: str, content_type: str, content_length: Optional[int] = ..., +) -> IO[bytes]: ... +def parse_form_data( + environ: WSGIEnvironment, + stream_factory: Optional[_StreamFactory] = ..., + charset: Text = ..., + errors: Text = ..., + max_form_memory_size: Optional[int] = ..., + max_content_length: Optional[int] = ..., + cls: Optional[Callable[[], _Dict]] = ..., + silent: bool = ..., +) -> Tuple[IO[bytes], _Dict, _Dict]: ... +def exhaust_stream(f: _F) -> _F: ... + +class FormDataParser(object): + stream_factory: _StreamFactory charset: Text errors: Text - max_form_memory_size: Any - max_content_length: Any - cls: Any - silent: Any - def __init__(self, stream_factory: Optional[Any] = ..., charset: Text = ..., errors: Text = ..., - max_form_memory_size: Optional[Any] = ..., max_content_length: Optional[Any] = ..., cls: Optional[Any] = ..., - silent: bool = ...): ... - def get_parse_func(self, mimetype, options): ... - def parse_from_environ(self, environ): ... - def parse(self, stream, mimetype, content_length, options: Optional[Any] = ...): ... - parse_functions: Any + max_form_memory_size: Optional[int] + max_content_length: Optional[int] + cls: Callable[[], _Dict] + silent: bool + def __init__( + self, + stream_factory: Optional[_StreamFactory] = ..., + charset: Text = ..., + errors: Text = ..., + max_form_memory_size: Optional[int] = ..., + max_content_length: Optional[int] = ..., + cls: Optional[Callable[[], _Dict]] = ..., + silent: bool = ..., + ) -> None: ... + def get_parse_func(self, mimetype: str, options: Any) -> Optional[_ParseFunc]: ... + def parse_from_environ(self, environ: WSGIEnvironment) -> Tuple[IO[bytes], _Dict, Dict]: ... + def parse( + self, stream: IO[bytes], mimetype: Text, content_length: Optional[int], options: Optional[Mapping[str, str]] = ..., + ) -> Tuple[IO[bytes], _Dict, _Dict]: ... + parse_functions: Dict[Text, _ParseFunc] -def is_valid_multipart_boundary(boundary): ... -def parse_multipart_headers(iterable): ... +def is_valid_multipart_boundary(boundary: str) -> bool: ... +def parse_multipart_headers(iterable: Iterable[Union[Text, bytes]]) -> Headers: ... -class MultiPartParser: +class MultiPartParser(object): charset: Text errors: Text - max_form_memory_size: Any - stream_factory: Any - cls: Any - buffer_size: Any - def __init__(self, stream_factory: Optional[Any] = ..., charset: Text = ..., errors: Text = ..., - max_form_memory_size: Optional[Any] = ..., cls: Optional[Any] = ..., buffer_size=...): ... - def fail(self, message): ... - def get_part_encoding(self, headers): ... - def get_part_charset(self, headers) -> Text: ... - def start_file_streaming(self, filename, headers, total_content_length): ... - def in_memory_threshold_reached(self, bytes): ... - def validate_boundary(self, boundary): ... - def parse_lines(self, file, boundary, content_length, cap_at_buffer: bool = ...): ... - def parse_parts(self, file, boundary, content_length): ... - def parse(self, file, boundary, content_length): ... + max_form_memory_size: Optional[int] + stream_factory: _StreamFactory + cls: Callable[[], _Dict] + buffer_size: int + def __init__( + self, + stream_factory: Optional[_StreamFactory] = ..., + charset: Text = ..., + errors: Text = ..., + max_form_memory_size: Optional[int] = ..., + cls: Optional[Callable[[], _Dict]] = ..., + buffer_size: int = ..., + ) -> None: ... + def fail(self, message: Text) -> NoReturn: ... + def get_part_encoding(self, headers: Mapping[str, str]) -> Optional[str]: ... + def get_part_charset(self, headers: Mapping[str, str]) -> Text: ... + def start_file_streaming( + self, filename: Union[Text, bytes], headers: Mapping[str, str], total_content_length: Optional[int], + ) -> Tuple[Text, IO[bytes]]: ... + def in_memory_threshold_reached(self, bytes: Any) -> NoReturn: ... + def validate_boundary(self, boundary: Optional[str]) -> None: ... + def parse_lines( + self, file: Any, boundary: bytes, content_length: int, cap_at_buffer: bool = ..., + ) -> Generator[Tuple[str, Any], None, None]: ... + def parse_parts(self, file: Any, boundary: bytes, content_length: int) -> Generator[Tuple[str, Any], None, None]: ... + def parse(self, file: Any, boundary: bytes, content_length: int) -> Tuple[_Dict, _Dict]: ...