Improve wsgiref stubs (#2145)

* Use typing.Text

* Add InputStream and ErrorStream WSGI protocols

* Fix return type of WSGIApplication

* Add type hints to wsgiref.validate

* Replace _Bytes by plain bytes

* Add wsgiref.util

* Add wsgiref.headers

* ErrorWrapper.writelines() takes an Iterable

* Fix start_response return type
* Make Python 2 and 3 branches resemble each other more closely
* Use typing.NoReturn
* Fix WriteWrapper.writer type

* Change return type of WriteWrapper.writer to Any

stdlib/2and3/wsgiref/headers.pyi

@@ -0,0 +1,31 @@
+import sys
+from typing import overload, Pattern, Optional, List, Tuple
+
+_HeaderList = List[Tuple[str, str]]
+
+tspecials: Pattern[str]  # undocumented
+
+class Headers:
+    if sys.version_info < (3, 5):
+        def __init__(self, headers: _HeaderList) -> None: ...
+    else:
+        def __init__(self, headers: Optional[_HeaderList] = ...) -> None: ...
+    def __len__(self) -> int: ...
+    def __setitem__(self, name: str, val: str) -> None: ...
+    def __delitem__(self, name: str) -> None: ...
+    def __getitem__(self, name: str) -> Optional[str]: ...
+    if sys.version_info < (3,):
+        def has_key(self, name: str) -> bool: ...
+    def __contains__(self, name: str) -> bool: ...
+    def get_all(self, name: str) -> List[str]: ...
+    @overload
+    def get(self, name: str, default: str) -> str: ...
+    @overload
+    def get(self, name: str, default: Optional[str] = ...) -> Optional[str]: ...
+    def keys(self) -> List[str]: ...
+    def values(self) -> List[str]: ...
+    def items(self) -> _HeaderList: ...
+    if sys.version_info >= (3,):
+        def __bytes__(self) -> bytes: ...
+    def setdefault(self, name: str, value: str) -> str: ...
+    def add_header(self, _name: str, _value: Optional[str], **_params: Optional[str]) -> None: ...

stdlib/2and3/wsgiref/types.pyi

@@ -1,7 +1,7 @@
 # Type declaration for a WSGI Function
 #
-# wsgiref/types.py doesn't exist and neither does WSGIApplication, it's a type
-# provided for type checking purposes.
+# wsgiref/types.py doesn't exist and neither do the types defined in this
+# file. They are provided for type checking purposes.
 #
 # This means you cannot simply import wsgiref.types in your code. Instead,
 # use the `TYPE_CHECKING` flag from the typing module:
@@ -15,27 +15,33 @@
 # you need to use 'WSGIApplication' and not simply WSGIApplication when type
 # hinting your code.  Otherwise Python will raise NameErrors.
 
-import sys
-from typing import Callable, Dict, Iterable, List, Optional, Tuple, Type, Union, Any
+from typing import Callable, Dict, Iterable, List, Optional, Tuple, Type, Union, Any, Text, Protocol
 from types import TracebackType
 
 _exc_info = Tuple[Optional[Type[BaseException]],
                   Optional[BaseException],
                   Optional[TracebackType]]
-if sys.version_info < (3,):
-    _Text = Union[unicode, str]
-    _BText = _Text
-else:
-    _Text = str
-    _BText = Union[bytes, str]
-WSGIEnvironment = Dict[_Text, Any]
+WSGIEnvironment = Dict[Text, Any]
 WSGIApplication = Callable[
     [
         WSGIEnvironment,
         Union[
-            Callable[[_Text, List[Tuple[_Text, _Text]]], Callable[[_BText], None]],
-            Callable[[_Text, List[Tuple[_Text, _Text]], _exc_info], Callable[[_BText], None]]
+            Callable[[Text, List[Tuple[Text, Text]]], Callable[[bytes], None]],
+            Callable[[Text, List[Tuple[Text, Text]], _exc_info], Callable[[bytes], None]]
         ]
     ],
-    Iterable[_BText]
+    Iterable[bytes]
 ]
+
+# WSGI input streams per PEP 3333
+class InputStream(Protocol):
+    def read(self, size: int = ...) -> bytes: ...
+    def readline(self, size: int = ...) -> bytes: ...
+    def readlines(self, hint: int = ...) -> List[bytes]: ...
+    def __iter__(self) -> Iterable[bytes]: ...
+
+# WSGI error streams per PEP 3333
+class ErrorStream(Protocol):
+    def flush(self) -> None: ...
+    def write(self, s: str) -> None: ...
+    def writelines(self, seq: List[str]) -> None: ...

stdlib/2and3/wsgiref/util.pyi

@@ -0,0 +1,23 @@
+import sys
+from typing import IO, Any, Optional
+
+from .types import WSGIEnvironment
+
+class FileWrapper:
+    filelike: IO[bytes]
+    blksize: int
+    def __init__(self, filelike: IO[bytes], bklsize: int = ...) -> None: ...
+    def __getitem__(self, key: Any) -> bytes: ...
+    def __iter__(self) -> FileWrapper: ...
+    if sys.version_info < (3,):
+        def next(self) -> bytes: ...
+    else:
+        def __next__(self) -> bytes: ...
+    def close(self) -> None: ...  # only exists if filelike.close exists
+
+def guess_scheme(environ: WSGIEnvironment) -> str: ...
+def application_uri(environ: WSGIEnvironment) -> str: ...
+def request_uri(environ: WSGIEnvironment, include_query: bool = ...) -> str: ...
+def shift_path_info(environ: WSGIEnvironment) -> Optional[str]: ...
+def setup_testing_defaults(environ: WSGIEnvironment) -> None: ...
+def is_hop_by_hop(header_name: str) -> bool: ...

stdlib/2and3/wsgiref/validate.pyi

@@ -1,50 +1,53 @@
 import sys
-from typing import Any
+from typing import Any, Iterable, Iterator, Optional, NoReturn, Callable
+
+from wsgiref.types import WSGIApplication, InputStream, ErrorStream
 
 class WSGIWarning(Warning): ...
 
-def validator(application): ...
+def validator(application: WSGIApplication) -> WSGIApplication: ...
 
 class InputWrapper:
-    input = ...  # type: Any
-    def __init__(self, wsgi_input): ...
-    def read(self, *args): ...
+    input: InputStream
+    def __init__(self, wsgi_input: InputStream) -> None: ...
     if sys.version_info < (3,):
-        def readline(self): ...
+        def read(self, size: int = ...) -> bytes: ...
+        def readline(self) -> bytes: ...
     else:
-        def readline(self, *args): ...
-    def readlines(self, *args): ...
-    def __iter__(self): ...
-    def close(self): ...
+        def read(self, size: int) -> bytes: ...
+        def readline(self, size: int = ...) -> bytes: ...
+    def readlines(self, hint: int = ...) -> bytes: ...
+    def __iter__(self) -> Iterable[bytes]: ...
+    def close(self) -> NoReturn: ...
 
 class ErrorWrapper:
-    errors = ...  # type: Any
-    def __init__(self, wsgi_errors): ...
-    def write(self, s): ...
-    def flush(self): ...
-    def writelines(self, seq): ...
-    def close(self): ...
+    errors: ErrorStream
+    def __init__(self, wsgi_errors: ErrorStream) -> None: ...
+    def write(self, s: str) -> None: ...
+    def flush(self) -> None: ...
+    def writelines(self, seq: Iterable[str]) -> None: ...
+    def close(self) -> NoReturn: ...
 
 class WriteWrapper:
-    writer = ...  # type: Any
-    def __init__(self, wsgi_writer): ...
-    def __call__(self, s): ...
+    writer: Callable[[bytes], Any]
+    def __init__(self, wsgi_writer: Callable[[bytes], Any]) -> None: ...
+    def __call__(self, s: bytes) -> None: ...
 
 class PartialIteratorWrapper:
-    iterator = ...  # type: Any
-    def __init__(self, wsgi_iterator): ...
-    def __iter__(self): ...
+    iterator: Iterator[bytes]
+    def __init__(self, wsgi_iterator: Iterator[bytes]) -> None: ...
+    def __iter__(self) -> IteratorWrapper: ...
 
 class IteratorWrapper:
-    original_iterator = ...  # type: Any
-    iterator = ...  # type: Any
-    closed = ...  # type: Any
-    check_start_response = ...  # type: Any
-    def __init__(self, wsgi_iterator, check_start_response): ...
-    def __iter__(self): ...
+    original_iterator: Iterator[bytes]
+    iterator: Iterator[bytes]
+    closed: bool
+    check_start_response: Optional[bool]
+    def __init__(self, wsgi_iterator: Iterator[bytes], check_start_response: Optional[bool]) -> None: ...
+    def __iter__(self) -> IteratorWrapper: ...
     if sys.version_info < (3,):
-        def next(self): ...
+        def next(self) -> bytes: ...
     else:
-        def __next__(self): ...
-    def close(self): ...
-    def __del__(self): ...
+        def __next__(self) -> bytes: ...
+    def close(self) -> None: ...
+    def __del__(self) -> None: ...