From da151609fae29ba7e811d83eb762b8c143379d8d Mon Sep 17 00:00:00 2001 From: Adam Johnson Date: Thu, 3 Nov 2022 08:38:56 +0000 Subject: [PATCH] many more types including wsgiref.types --- src/whitenoise/base.py | 32 ++++++++++----- src/whitenoise/compat.py | 73 +++++++++++++++++++++++++++++++++ src/whitenoise/responders.py | 5 ++- tests/test_django_whitenoise.py | 5 ++- tests/utils.py | 12 ++++-- 5 files changed, 109 insertions(+), 18 deletions(-) create mode 100644 src/whitenoise/compat.py diff --git a/src/whitenoise/base.py b/src/whitenoise/base.py index 45dcc1db..f1edcb14 100644 --- a/src/whitenoise/base.py +++ b/src/whitenoise/base.py @@ -4,10 +4,11 @@ import re import warnings from posixpath import normpath -from typing import Callable, Generator +from typing import Callable, Generator, Iterable from wsgiref.headers import Headers from wsgiref.util import FileWrapper +from .compat import StartResponse, WSGIApplication, WSGIEnvironment from .media_types import MediaTypes from .responders import IsDirectoryError, MissingFileError, Redirect, StaticFile from .string_utils import ( @@ -25,7 +26,7 @@ class WhiteNoise: def __init__( self, - application, + application: WSGIApplication, root: str | None = None, prefix: str | None = None, *, @@ -65,12 +66,14 @@ def __init__( self.media_types = MediaTypes(extra_types=mimetypes) self.application = application - self.files = {} - self.directories = [] + self.files: dict[str, Redirect | StaticFile] = {} + self.directories: list[tuple[str, str]] = [] if root is not None: self.add_files(root, prefix) - def __call__(self, environ, start_response): + def __call__( + self, environ: WSGIEnvironment, start_response: StartResponse + ) -> Iterable[bytes]: path = decode_path_info(environ.get("PATH_INFO", "")) if self.autorefresh: static_file = self.find_file(path) @@ -82,12 +85,18 @@ def __call__(self, environ, start_response): return self.serve(static_file, environ, start_response) @staticmethod - def serve(static_file, environ, start_response): + def serve( + static_file: Redirect | StaticFile, + environ: WSGIEnvironment, + start_response: StartResponse, + ) -> Iterable[bytes]: response = static_file.get_response(environ["REQUEST_METHOD"], environ) status_line = f"{response.status} {response.status.phrase}" start_response(status_line, list(response.headers)) if response.file is not None: - file_wrapper = environ.get("wsgi.file_wrapper", FileWrapper) + file_wrapper: type[FileWrapper] = environ.get( + "wsgi.file_wrapper", FileWrapper + ) return file_wrapper(response.file) else: return [] @@ -139,14 +148,15 @@ def add_file_to_dictionary( def find_file(self, url: str) -> Redirect | StaticFile | None: # Optimization: bail early if the URL can never match a file if not self.index_file and url.endswith("/"): - return + return None if not self.url_is_canonical(url): - return + return None for path in self.candidate_paths_for_url(url): try: return self.find_file_at_path(path, url) except MissingFileError: pass + return None def candidate_paths_for_url(self, url: str) -> Generator[str, None, None]: for root, prefix in self.directories: @@ -181,7 +191,7 @@ def find_file_at_path_with_indexes( raise MissingFileError(path) @staticmethod - def url_is_canonical(url): + def url_is_canonical(url: str) -> bool: """ Check that the URL path is in canonical format i.e. has normalised slashes and no path traversal elements @@ -195,7 +205,7 @@ def url_is_canonical(url): @staticmethod def is_compressed_variant( - path, stat_cache: dict[str, os.stat_result] | None = None + path: str, stat_cache: dict[str, os.stat_result] | None = None ) -> bool: if path[-3:] in (".gz", ".br"): uncompressed_path = path[:-3] diff --git a/src/whitenoise/compat.py b/src/whitenoise/compat.py new file mode 100644 index 00000000..260f7bcf --- /dev/null +++ b/src/whitenoise/compat.py @@ -0,0 +1,73 @@ +from __future__ import annotations + +import sys + +if sys.version_info >= (3, 11): + from wsgiref.types import WSGIApplication +else: + from collections.abc import Callable, Iterable, Iterator + from types import TracebackType + from typing import Any, Dict, Protocol, Tuple, Type, Union + + from typing_extensions import TypeAlias + + __all__ = [ + "StartResponse", + "WSGIEnvironment", + "WSGIApplication", + "InputStream", + "ErrorStream", + "FileWrapper", + ] + + _ExcInfo: TypeAlias = Tuple[Type[BaseException], BaseException, TracebackType] + _OptExcInfo: TypeAlias = Union[_ExcInfo, Tuple[None, None, None]] + + class StartResponse(Protocol): + def __call__( + self, + __status: str, + __headers: list[tuple[str, str]], + __exc_info: _OptExcInfo | None = ..., + ) -> Callable[[bytes], object]: + ... + + WSGIEnvironment: TypeAlias = Dict[str, Any] + WSGIApplication: TypeAlias = Callable[ + [WSGIEnvironment, StartResponse], Iterable[bytes] + ] + + 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) -> Iterator[bytes]: + ... + + class ErrorStream(Protocol): + def flush(self) -> object: + ... + + def write(self, __s: str) -> object: + ... + + def writelines(self, __seq: list[str]) -> object: + ... + + class _Readable(Protocol): + def read(self, __size: int = ...) -> bytes: + ... + + # Optional: def close(self) -> object: ... + + class FileWrapper(Protocol): + def __call__( + self, __file: _Readable, __block_size: int = ... + ) -> Iterable[bytes]: + ... diff --git a/src/whitenoise/responders.py b/src/whitenoise/responders.py index a498a3ce..2acd65af 100644 --- a/src/whitenoise/responders.py +++ b/src/whitenoise/responders.py @@ -17,7 +17,10 @@ class Response: __slots__ = ("status", "headers", "file") def __init__( - self, status: int, headers: Sequence[tuple[str, str]], file: BinaryIO | None + self, + status: HTTPStatus, + headers: Sequence[tuple[str, str]], + file: BinaryIO | None, ) -> None: self.status = status self.headers = headers diff --git a/tests/test_django_whitenoise.py b/tests/test_django_whitenoise.py index 6a7dbdf8..c811e66b 100644 --- a/tests/test_django_whitenoise.py +++ b/tests/test_django_whitenoise.py @@ -3,6 +3,7 @@ import shutil import tempfile from contextlib import closing +from typing import Any from urllib.parse import urljoin, urlparse import pytest @@ -18,11 +19,11 @@ from .utils import AppServer, Files -def reset_lazy_object(obj): +def reset_lazy_object(obj: Any) -> None: obj._wrapped = empty -def get_url_path(base, url): +def get_url_path(base: str, url: str) -> str: return urlparse(urljoin(base, url)).path diff --git a/tests/utils.py b/tests/utils.py index 639e899c..c4ce8439 100644 --- a/tests/utils.py +++ b/tests/utils.py @@ -2,17 +2,19 @@ import os import threading -from typing import Any +from typing import Any, Iterable from wsgiref.simple_server import WSGIRequestHandler, make_server from wsgiref.util import shift_path_info import requests +from whitenoise.compat import StartResponse, WSGIApplication, WSGIEnvironment + TEST_FILE_PATH = os.path.join(os.path.dirname(__file__), "test_files") class SilentWSGIHandler(WSGIRequestHandler): - def log_message(*args): + def log_message(self, format: str, *args: Any) -> None: pass @@ -24,13 +26,15 @@ class AppServer: PREFIX = "subdir" - def __init__(self, application): + def __init__(self, application: WSGIApplication) -> None: self.application = application self.server = make_server( "127.0.0.1", 0, self.serve_under_prefix, handler_class=SilentWSGIHandler ) - def serve_under_prefix(self, environ, start_response): + def serve_under_prefix( + self, environ: WSGIEnvironment, start_response: StartResponse + ) -> Iterable[bytes]: prefix = shift_path_info(environ) if prefix != self.PREFIX: start_response("404 Not Found", [])