Add types to FileStorage (#3067)

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]: ...

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]: ...