Add Types

src/app_server/__init__.py

@@ -49,12 +49,16 @@ def start_server(
             app.wsgi_app, {route["url"]: os.path.join(app_folder, path)}
         )
 
-    apps["/"] = Proxy(app.wsgi_app, {
-        "/": {
-            "target": f"{protocol}://{host}:{gunicorn_port}/",
-            "host": None
-        }
-    }, timeout=timeout)
+    apps["/"] = Proxy(
+        app.wsgi_app,
+        {
+            "/": {
+                "target": f"{protocol}://{host}:{gunicorn_port}/",
+                "host": None
+            }
+        },
+        timeout=timeout
+    )
     app.wsgi_app = Dispatcher(app.wsgi_app, apps)
 
     run_simple(host, port, app, use_debugger=False, use_reloader=True,

src/app_server/app_wrapper.py

@@ -1,12 +1,22 @@
+import typing as t
+from wsgiref.types import StartResponse, WSGIEnvironment
+
 from werkzeug.wrappers import Request, Response
+
+
 class AppWrapper:
     """simple wrapping app"""
+
     @staticmethod
-    def wsgi_app(environ, start_response):
+    def wsgi_app(
+        environ: WSGIEnvironment, start_response: StartResponse
+    ) -> t.Iterable[bytes]:
         request = Request(environ)
         response = Response(f'Path not found or invalid: {request.path}',
                             status=404)
         return response(environ, start_response)
 
-    def __call__(self, environ, start_response):
+    def __call__(
+        self, environ: WSGIEnvironment, start_response: StartResponse
+    ) -> t.Iterable[bytes]:
         return self.wsgi_app(environ, start_response)

src/app_server/dispatcher.py

@@ -1,11 +1,16 @@
 import re
+import typing as t
+from wsgiref.types import StartResponse, WSGIEnvironment
+
 from werkzeug.middleware.dispatcher import DispatcherMiddleware
 
 
 class Dispatcher(DispatcherMiddleware):
     """use regex to find a matching route"""
 
-    def __call__(self, environ, start_response):
+    def __call__(
+        self, environ: WSGIEnvironment, start_response: StartResponse
+    ) -> t.Iterable[bytes]:
         app = self.mounts["/"]
 
         for route, _app in self.mounts.items():

src/app_server/proxy.py

@@ -1,11 +1,20 @@
+import typing as t
+from wsgiref.types import StartResponse, WSGIApplication, WSGIEnvironment
+
 from werkzeug.middleware.http_proxy import ProxyMiddleware
 from werkzeug.wsgi import get_path_info
-import typing as t
+
 
 class Proxy(ProxyMiddleware):
     """this addition allows to redirect all routes to given targets"""
 
-    def __init__(self, app, targets, chunk_size=2 << 13, timeout=10):
+    def __init__(
+        self,
+        app: WSGIApplication,
+        targets: t.Mapping[str, dict[str, t.Any]],
+        chunk_size: int = 2 << 13,
+        timeout: int = 10,
+    ) -> None:
         super().__init__(app, targets, chunk_size, timeout)
 
         def _set_defaults(opts):
@@ -19,8 +28,9 @@ def _set_defaults(opts):
             f"{k}": _set_defaults(v) for k, v in targets.items()
         }
 
-    def __call__(self, environ: "WSGIEnvironment",
-                 start_response: "StartResponse") -> t.Iterable[bytes]:
+    def __call__(
+        self, environ: WSGIEnvironment, start_response: StartResponse
+    ) -> t.Iterable[bytes]:
         path = get_path_info(environ, charset='utf-8', errors='replace')
         app = self.app
         for prefix, opts in self.targets.items():

src/app_server/request_handler.py

@@ -2,15 +2,16 @@
 import typing as t
 
 from werkzeug._internal import _logger  # noqa
-from werkzeug.serving import WSGIRequestHandler , _ansi_style, \
+from werkzeug.serving import WSGIRequestHandler, _ansi_style, \
     _log_add_style
 from werkzeug.urls import uri_to_iri
 
 import logging
 
 
 class CustomWSGIRequestHandler(WSGIRequestHandler):
-    def log_date_time_string(self):
+    @staticmethod
+    def log_date_time_string():
         """Return the current time formatted for logging."""
         now = time.time()
         year, month, day, hh, mm, ss, x, y, z = time.localtime(now)

src/app_server/shared_data.py

@@ -1,6 +1,9 @@
+import mimetypes
 import re
 import time
-import mimetypes
+import typing as t
+from wsgiref.types import StartResponse, WSGIApplication, WSGIEnvironment
+
 from werkzeug.http import http_date, is_resource_modified
 from werkzeug.middleware.shared_data import SharedDataMiddleware
 from werkzeug.utils import get_content_type
@@ -12,18 +15,24 @@ class SharedData(SharedDataMiddleware):
 
     def __init__(
         self,
-        app,
-        exports,
+        app: WSGIApplication,
+        exports: (
+            dict[str, str | tuple[str, str]]
+            | t.Iterable[tuple[str, str | tuple[str, str]]]
+        ),
         disallow: None = None,
         cache: bool = True,
         cache_timeout: int = 60 * 60 * 12,
         fallback_mimetype: str = "application/octet-stream",
     ) -> None:
+
         self.org_exports = exports.copy()
         super().__init__(app, exports, disallow, cache, cache_timeout,
                          fallback_mimetype)
 
-    def __call__(self, environ, start_response):
+    def __call__(
+        self, environ: WSGIEnvironment, start_response: StartResponse
+    ) -> t.Iterable[bytes]:
         path = get_path_info(environ)
         file_loader = None
 

src/app_server/utils.py

@@ -17,6 +17,7 @@ def patch_gunicorn():
             '        except (SyntaxError, NameError) as e:'
         ))
 
+
 def set_env_vars(application_id: str, args: argparse.Namespace, app_yaml: dict):
     """set necessary environment variables"""
     # First, merge the app.yaml into the environment so that the variables