From 8b84e9cf13ee20afd801e3d03b7a6700fea69222 Mon Sep 17 00:00:00 2001 From: Sebastian Rittau Date: Wed, 23 May 2018 17:19:07 +0200 Subject: [PATCH] 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 | 31 ++++++++++++++ stdlib/2and3/wsgiref/types.pyi | 34 +++++++++------- stdlib/2and3/wsgiref/util.pyi | 23 +++++++++++ stdlib/2and3/wsgiref/validate.pyi | 67 ++++++++++++++++--------------- 4 files changed, 109 insertions(+), 46 deletions(-) create mode 100644 stdlib/2and3/wsgiref/headers.pyi create mode 100644 stdlib/2and3/wsgiref/util.pyi diff --git a/stdlib/2and3/wsgiref/headers.pyi b/stdlib/2and3/wsgiref/headers.pyi new file mode 100644 index 000000000..853922527 --- /dev/null +++ b/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: ... diff --git a/stdlib/2and3/wsgiref/types.pyi b/stdlib/2and3/wsgiref/types.pyi index 172028123..2dc096411 100644 --- a/stdlib/2and3/wsgiref/types.pyi +++ b/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: ... diff --git a/stdlib/2and3/wsgiref/util.pyi b/stdlib/2and3/wsgiref/util.pyi new file mode 100644 index 000000000..501ddcd5b --- /dev/null +++ b/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: ... diff --git a/stdlib/2and3/wsgiref/validate.pyi b/stdlib/2and3/wsgiref/validate.pyi index a12daf421..c7fa699fc 100644 --- a/stdlib/2and3/wsgiref/validate.pyi +++ b/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: ...