add type hints to all of the code

In order to use type hint syntax, we need to bump the minimum Python
version to 3.7 and some of the features such as Literal and Protocol are
opted in when a sufficiently recent Python is available.

This does not make all of the code pass type checking with mypy. A
number of typing issues remain, but the output of mypy becomes something
one can read through.

In adding type hints, a lot of epydoc @type annotations are removed as
redundant. This update also adopts black-style line breaking.

1 files changed, 107 insertions, 69 deletions

diff --git a/wsgitools/middlewares.py b/wsgitools/middlewares.py
index ef9fe84..8577384 100644
--- a/wsgitools/middlewares.py
+++ b/wsgitools/middlewares.py
@@ -6,27 +6,28 @@ import sys
 import cgitb
 import collections
 import io
+import typing
 
 from wsgitools.internal import bytes2str, str2bytes
 from wsgitools.filters import CloseableList, CloseableIterator
 from wsgitools.authentication import AuthenticationRequired, \
         ProtocolViolation, AuthenticationMiddleware
+from wsgitools.internal import (
+    Environ, HeaderList, OptExcInfo, StartResponse, WriteCallback, WsgiApp
+)
 
 __all__.append("SubdirMiddleware")
 class SubdirMiddleware:
     """Middleware choosing wsgi applications based on a dict."""
-    def __init__(self, default, mapping={}):
-        """
-        @type default: wsgi app
-        @type mapping: {str: wsgi app}
-        """
+    def __init__(
+        self, default: WsgiApp, mapping: typing.Dict[str, WsgiApp] = {}
+    ):
         self.default = default
         self.mapping = mapping
-    def __call__(self, environ, start_response):
-        """wsgi interface
-        @type environ: {str: str}
-        @rtype: gen([bytes])
-        """
+    def __call__(
+        self, environ: Environ, start_response: StartResponse
+    ) -> typing.Iterable[bytes]:
+        """wsgi interface"""
         assert isinstance(environ, dict)
         app = None
         script = environ["PATH_INFO"]
@@ -51,22 +52,24 @@ class NoWriteCallableMiddleware:
     C{start_response} function to a wsgi application that doesn't need one by
     writing the data to a C{BytesIO} and then making it be the first result
     element."""
-    def __init__(self, app):
+    def __init__(self, app: WsgiApp):
         """Wraps wsgi application app."""
         self.app = app
-    def __call__(self, environ, start_response):
-        """wsgi interface
-        @type environ: {str, str}
-        @rtype: gen([bytes])
-        """
+    def __call__(
+        self, environ: Environ, start_response: StartResponse
+    ) -> typing.Iterable[bytes]:
+        """wsgi interface"""
         assert isinstance(environ, dict)
-        todo = [None]
+        todo: typing.Optional[typing.Tuple[str, HeaderList]] = None
         sio = io.BytesIO()
         gotiterdata = False
-        def write_calleable(data):
+        def write_calleable(data: bytes) -> None:
             assert not gotiterdata
             sio.write(data)
-        def modified_start_response(status, headers, exc_info=None):
+        def modified_start_response(
+            status: str, headers: HeaderList, exc_info: OptExcInfo = None
+        ) -> WriteCallback:
+            nonlocal todo
             try:
                 if sio.tell() > 0 or gotiterdata:
                     assert exc_info is not None
@@ -75,7 +78,7 @@ class NoWriteCallableMiddleware:
                 exc_info = None
             assert isinstance(status, str)
             assert isinstance(headers, list)
-            todo[0] = (status, headers)
+            todo = (status, headers)
             return write_calleable
 
         ret = self.app(environ, modified_start_response)
@@ -96,8 +99,8 @@ class NoWriteCallableMiddleware:
         else:
             gotiterdata = True
 
-        assert todo[0] is not None
-        status, headers = todo[0]
+        assert todo is not None
+        status, headers = todo
         data = sio.getvalue()
 
         if isinstance(ret, list):
@@ -117,27 +120,36 @@ class ContentLengthMiddleware:
     """Guesses the content length header if possible.
     @note: The application used must not use the C{write} callable returned by
            C{start_response}."""
-    def __init__(self, app, maxstore=0):
+    maxstore: typing.Union[float, int]
+    def __init__(
+        self, app: WsgiApp, maxstore: typing.Union[int, typing.Tuple[()]] = 0
+    ):
         """Wraps wsgi application app. If the application returns a list, the
         total length of strings is available and the content length header is
         set unless there already is one. For an iterator data is accumulated up
         to a total of maxstore bytes (where maxstore=() means infinity). If the
         iterator is exhaused within maxstore bytes a content length header is
         added unless already present.
-        @type maxstore: int or ()
         @note: that setting maxstore to a value other than 0 will violate the
                wsgi standard
         """
         self.app = app
         if maxstore == ():
-            maxstore = float("inf")
-        self.maxstore = maxstore
-    def __call__(self, environ, start_response):
+            self.maxstore = float("inf")
+        else:
+            assert isinstance(maxstore, int)
+            self.maxstore = maxstore
+    def __call__(
+        self, environ: Environ, start_response: StartResponse
+    ) -> typing.Iterable[bytes]:
         """wsgi interface"""
         assert isinstance(environ, dict)
-        todo = []
+        todo: typing.Optional[typing.Tuple[str, HeaderList]] = None
         gotdata = False
-        def modified_start_response(status, headers, exc_info=None):
+        def modified_start_response(
+            status: str, headers: HeaderList, exc_info: OptExcInfo = None
+        ) -> WriteCallback:
+            nonlocal todo
             try:
                 if gotdata:
                     assert exc_info is not None
@@ -146,8 +158,8 @@ class ContentLengthMiddleware:
                 exc_info = None
             assert isinstance(status, str)
             assert isinstance(headers, list)
-            todo[:] = ((status, headers),)
-            def raise_not_imp(*args):
+            todo = (status, headers)
+            def raise_not_imp(_: bytes) -> None:
                 raise NotImplementedError
             return raise_not_imp
 
@@ -156,8 +168,8 @@ class ContentLengthMiddleware:
 
         if isinstance(ret, list):
             gotdata = True
-            assert bool(todo)
-            status, headers = todo[0]
+            assert todo is not None
+            status, headers = todo
             if all(k.lower() != "content-length" for k, _ in headers):
                 length = sum(map(len, ret))
                 headers.append(("Content-Length", str(length)))
@@ -173,8 +185,8 @@ class ContentLengthMiddleware:
             except StopIteration:
                 stopped = True
         gotdata = True
-        assert bool(todo)
-        status, headers = todo[0]
+        assert todo is not None
+        status, headers = todo
         data = CloseableList(getattr(ret, "close", None))
         if first:
             data.append(first)
@@ -197,12 +209,12 @@ class ContentLengthMiddleware:
 
         return CloseableIterator(getattr(ret, "close", None), data, ret)
 
-def storable(environ):
+def storable(environ: Environ) -> bool:
     if environ["REQUEST_METHOD"] != "GET":
         return False
     return True
 
-def cacheable(environ):
+def cacheable(environ: Environ) -> bool:
     if environ.get("HTTP_CACHE_CONTROL", "") == "max-age=0":
         return False
     return True
@@ -213,16 +225,24 @@ class CachingMiddleware:
     C{QUERY_STRING}."""
 
     class CachedRequest:
-        def __init__(self, timestamp):
+        def __init__(self, timestamp: float):
             self.timestamp = timestamp
             self.status = ""
-            self.headers = []
-            self.body = []
-
-    def __init__(self, app, maxage=60, storable=storable, cacheable=cacheable):
+            self.headers: HeaderList = []
+            self.body: typing.List[bytes] = []
+
+    cache: typing.Dict[str, CachedRequest]
+    lastcached: typing.Deque[typing.Tuple[str, float]]
+
+    def __init__(
+        self,
+        app: WsgiApp,
+        maxage: int = 60,
+        storable: typing.Callable[[Environ], bool] = storable,
+        cacheable: typing.Callable[[Environ], bool] = cacheable,
+    ):
         """
         @param app: is a wsgi application to be cached.
-        @type maxage: int
         @param maxage: is the number of seconds a reponse may be cached.
         @param storable: is a predicate that determines whether the response
                 may be cached at all based on the C{environ} dict.
@@ -235,13 +255,20 @@ class CachingMiddleware:
         self.cache = {}
         self.lastcached = collections.deque()
 
-    def insert_cache(self, key, obj, now=None):
+    def insert_cache(
+        self,
+        key: str,
+        obj: CachedRequest,
+        now: typing.Optional[float] = None,
+    ) -> None:
         if now is None:
             now = time.time()
         self.cache[key] = obj
         self.lastcached.append((key, now))
 
-    def prune_cache(self, maxclean=16, now=None):
+    def prune_cache(
+        self, maxclean: int = 16, now: typing.Optional[float] = None
+    ) -> None:
         if now is None:
             now = time.time()
         old = now - self.maxage
@@ -258,10 +285,10 @@ class CachingMiddleware:
                 if obj.timestamp <= old:
                     del self.cache[key]
 
-    def __call__(self, environ, start_response):
-        """wsgi interface
-        @type environ: {str: str}
-        """
+    def __call__(
+        self, environ: Environ, start_response: StartResponse
+    ) -> typing.Iterable[bytes]:
+        """wsgi interface"""
         assert isinstance(environ, dict)
         now = time.time()
         self.prune_cache(now=now)
@@ -279,7 +306,9 @@ class CachingMiddleware:
             else:
                 del self.cache[path]
         cache_object = self.CachedRequest(now)
-        def modified_start_respesponse(status, headers, exc_info=None):
+        def modified_start_respesponse(
+            status: str, headers: HeaderList, exc_info: OptExcInfo = None
+        ) -> WriteCallback:
             try:
                 if cache_object.body:
                     assert exc_info is not None
@@ -291,7 +320,7 @@ class CachingMiddleware:
             cache_object.status = status
             cache_object.headers = headers
             write = start_response(status, list(headers))
-            def modified_write(data):
+            def modified_write(data: bytes) -> None:
                 cache_object.body.append(data)
                 write(data)
             return modified_write
@@ -303,7 +332,7 @@ class CachingMiddleware:
             cache_object.body.extend(ret)
             self.insert_cache(path, cache_object, now)
             return ret
-        def pass_through():
+        def pass_through() -> typing.Iterator[bytes]:
             for data in ret:
                 cache_object.body.append(data)
                 yield data
@@ -313,18 +342,13 @@ class CachingMiddleware:
 __all__.append("DictAuthChecker")
 class DictAuthChecker:
     """Verifies usernames and passwords by looking them up in a dict."""
-    def __init__(self, users):
+    def __init__(self, users: typing.Dict[str, str]):
         """
-        @type users: {str: str}
         @param users: is a dict mapping usernames to password."""
         self.users = users
-    def __call__(self, username, password, environ):
+    def __call__(self, username: str, password: str, environ: Environ) -> bool:
         """check_function interface taking username and password and resulting
         in a bool.
-        @type username: str
-        @type password: str
-        @type environ: {str: object}
-        @rtype: bool
         """
         return username in self.users and self.users[username] == password
 
@@ -334,7 +358,13 @@ class BasicAuthMiddleware(AuthenticationMiddleware):
     the warpped application the environ dictionary is augmented by a REMOTE_USER
     key."""
     authorization_method = "basic"
-    def __init__(self, app, check_function, realm='www', app401=None):
+    def __init__(
+        self,
+        app: WsgiApp,
+        check_function: typing.Callable[[str, str, Environ], bool],
+        realm: str = 'www',
+        app401: typing.Optional[WsgiApp] = None,
+    ):
         """
         @param app: is a WSGI application.
         @param check_function: is a function taking three arguments username,
@@ -342,7 +372,6 @@ class BasicAuthMiddleware(AuthenticationMiddleware):
                 request may is allowed. The older interface of taking only the
                 first two arguments is still supported via catching a
                 C{TypeError}.
-        @type realm: str
         @param app401: is an optional WSGI application to be used for error
                 messages
         """
@@ -351,7 +380,9 @@ class BasicAuthMiddleware(AuthenticationMiddleware):
         self.realm = realm
         self.app401 = app401
 
-    def authenticate(self, auth, environ):
+    def authenticate(
+        self, auth: str, environ: Environ
+    ) -> typing.Dict[str, str]:
         assert isinstance(auth, str)
         assert isinstance(environ, dict)
         authb = str2bytes(auth)
@@ -372,10 +403,17 @@ class BasicAuthMiddleware(AuthenticationMiddleware):
             return dict(user=username)
         raise AuthenticationRequired("credentials not valid")
 
-    def www_authenticate(self, exception):
+    def www_authenticate(
+        self, exception: AuthenticationRequired
+    ) -> typing.Tuple[str, str]:
         return ("WWW-Authenticate", 'Basic realm="%s"' % self.realm)
 
-    def authorization_required(self, environ, start_response, exception):
+    def authorization_required(
+        self,
+        environ: Environ,
+        start_response: StartResponse,
+        exception: AuthenticationRequired,
+    ) -> typing.Iterable[bytes]:
         if self.app401 is not None:
             return self.app401(environ, start_response)
         return AuthenticationMiddleware.authorization_required(
@@ -385,13 +423,13 @@ __all__.append("TracebackMiddleware")
 class TracebackMiddleware:
     """In case the application throws an exception this middleware will show an
     html-formatted traceback using C{cgitb}."""
-    def __init__(self, app):
+    def __init__(self, app: WsgiApp):
         """app is the wsgi application to proxy."""
         self.app = app
-    def __call__(self, environ, start_response):
-        """wsgi interface
-        @type environ: {str: str}
-        """
+    def __call__(
+        self, environ: Environ, start_response: StartResponse
+    ) -> typing.Iterable[bytes]:
+        """wsgi interface"""
         try:
             assert isinstance(environ, dict)
             ret = self.app(environ, start_response)