Merge pull request #18 from the-scouts/typing-mypy

environment.yml

@@ -3,7 +3,7 @@ channels:
   - conda-forge
   - defaults
 dependencies:
-  - python>=3.7
+  - python>=3.9
 
   # dash packages
   - dash
@@ -16,12 +16,15 @@ dependencies:
   - pyarrow
   - python-dateutil
   - redis-py
+  - pydantic
 
   # development
   - black
   - pytest
   - toml
   - isort
+  - mypy
+  - ipython
 #
 #  # ci
 #  - gunicorn

pyproject.toml

@@ -1,29 +1,15 @@
 [tool.black]
 line-length = 180
-exclude = '''
-/(
-    \.eggs
-  | \.git
-  | \.hg
-  | \.mypy_cache
-  | \.tox
-  | \.venv
-  | venv
-  | _build
-  | buck-out
-  | build
-  | dist
-  | setup.py
-)/
-'''
+target-version = ["py38"]  # Remember to change this if we change pins from Python 3.9!  TODO black does not yet support 3.9 (13-01-2021)
 
 [tool.isort]
 line_length = 180
-skip_glob = '\.eggs/*,\.git/*,\.venv/*,build/*,dist/*'
+profile = "google"
+py_version = "39"
+skip_gitignore = true
 no_lines_before = "STDLIB"
 single_line_exclusions = ["dash.dependencies", "typing", "dependencies"]
-# force_single_line = "True" wait until isort 5
 known_dash = ["dash", "dash_core_components", "dash_html_components", "dash.dependencies"]
 known_pandas = ["pandas", "numpy", "pyarrow"]
 known_typing = "typing"
-sections = ["FUTURE","STDLIB","THIRDPARTY","DASH","PANDAS","FIRSTPARTY","LOCALFOLDER","TYPING"]
+sections = ["FUTURE", "STDLIB", "THIRDPARTY", "DASH"," PANDAS", "FIRSTPARTY", "LOCALFOLDER", "TYPING"]

setup.cfg

@@ -0,0 +1,55 @@
+[mypy]
+plugins = pydantic.mypy
+# help finding errors
+show_error_codes = True
+show_error_context = True
+# suppress errors from imported modules
+follow_imports = silent
+# disable re-exporting names
+no_implicit_reexport = True
+# warn about config that doesn't hit files
+warn_unused_configs = True
+# mypy --strict
+disallow_any_generics = True
+disallow_subclassing_any = True
+#disallow_untyped_calls = True
+disallow_untyped_defs = True
+disallow_incomplete_defs = True
+check_untyped_defs = True
+#disallow_untyped_decorators = True
+no_implicit_optional = True
+warn_redundant_casts = True
+warn_unused_ignores = True
+warn_return_any = True
+implicit_reexport = False
+strict_equality = True
+
+[pydantic-mypy]
+init_forbid_extra = True
+init_typed = True
+warn_required_dynamic_aliases = True
+warn_untyped_fields = True
+
+[mypy-dash_core_components]
+ignore_missing_imports = true
+
+[mypy-dash_html_components]
+ignore_missing_imports = true
+
+[mypy-dash]
+ignore_missing_imports = true
+
+[mypy-dash.*]
+ignore_missing_imports = true
+
+[mypy-lxml]
+ignore_missing_imports = true
+
+[mypy-pandas]
+ignore_missing_imports = true
+
+[mypy-pyarrow]
+ignore_missing_imports = true
+
+[mypy-flask_caching]
+ignore_missing_imports = true

src/auth.py

@@ -1,18 +1,25 @@
+from collections.abc import Callable
 import functools
 
 from flask import session
 
 import dash
+from dash.dependencies import Input, Output, State
 import dash_core_components as dcc
 import dash_html_components as html
-from dash.dependencies import Input, Output, State
 
-from src.config import password, user
+from src.config import password
+from src.config import user
+
+from typing import Any, Optional, TypeVar, Union
 
 users = {user: password}
 
+# https://mypy.readthedocs.io/en/stable/generics.html#declaring-decorators
+F = TypeVar('F', bound=Callable[..., Any])
+
 
-def authenticate_user(credentials):
+def authenticate_user(credentials: tuple[Optional[str], Optional[str]]) -> bool:
     """
     generic authentication function
     returns True if user is correct and False otherwise
@@ -26,14 +33,15 @@ def authenticate_user(credentials):
     return authed
 
 
-def validate_login_session(f):
+def validate_login_session(f: F) -> Union[F, html.Div]:
     """
     takes a layout function that returns layout objects
     checks if the user is logged in or not through the session.
     If not, returns an error with link to the login page
     """
     @functools.wraps(f)
-    def wrapper(*args, **kwargs):
+    def wrapper(*args: Any, **kwargs: Any) -> Union[F, html.Div]:
+        # return f(*args, **kwargs)
         if session.get("authed"):
             return f(*args, **kwargs)
         return html.Div(
@@ -48,7 +56,7 @@ def wrapper(*args, **kwargs):
 
 
 # login layout content
-def login_layout():
+def login_layout() -> html.Div:
     return html.Div(
         html.Div([
             html.H4("Login"),
@@ -62,7 +70,7 @@ def login_layout():
     )
 
 
-def setup_callbacks(app: dash.Dash):
+def setup_callbacks(app: dash.Dash) -> None:
     # authenticate
     @app.callback(
         [Output("login-url", "pathname"),
@@ -72,7 +80,7 @@ def setup_callbacks(app: dash.Dash):
          State("login-password", "value"),
          State("old-url", "data"), ],
         prevent_initial_call=True)
-    def login_auth(_, username, pw, pathname):
+    def login_auth(_: int, username: Optional[str], pw: Optional[str], pathname: str) -> Union[tuple[dash.no_update, html.Div], tuple[str, str]]:
         """
         check credentials
         if correct, authenticate the session
@@ -82,7 +90,7 @@ def login_auth(_, username, pw, pathname):
             pathname = "/home"
 
         credentials = (username, pw)
-        if credentials.count(None) == len(credentials):
+        if credentials == (None, None):
             return dash.no_update, dash.no_update
 
         if authenticate_user(credentials):

src/cache.py

@@ -1,12 +1,15 @@
 import datetime
 import json
 
-import dash
 import flask_caching
 import redis
 
+import dash
+
 import src.config as config
 
+from typing import Any, Optional
+
 
 class CacheInterface:
     cache_path = config.DATA_ROOT / "cache" / "cache.json"
@@ -27,55 +30,55 @@ def __init__(self, app: dash.Dash):
         self.environment = "prod" if config.is_production else "dev"
         self.app = app
 
-    def _serialize_path(self, *path) -> str:
+    def _serialize_path(self, *path: str) -> str:
         return "/".join([self.environment, *path])
 
     @staticmethod
-    def _set_by_path(root: dict, items, value):
+    def _set_by_path(root: dict[str, Any], items: list[str], value: Any) -> None:
         """Set a value in a nested object in root by item sequence."""
         for key in items[:-1]:
             root.setdefault(key, {})
             root = root[key]
         root[items[-1]] = value
 
     @staticmethod
-    def _keys_to_strings(keys: list, key_prefix: str = "") -> list:
-        return [key.decode("UTF8")[len(key_prefix):] for key in keys]  # TODO key.removeprefix(root) when py3.9 is released (Oct 20)
+    def _keys_to_strings(keys: list[bytes], key_prefix: str = "") -> list[str]:
+        return [key.decode("UTF8").removeprefix(key_prefix) for key in keys]
 
-    def get_keys_from_partial(self, *path) -> list:
+    def get_keys_from_partial(self, *path: str) -> list[bytes]:
         """Return a list of keys from the global cache"""
         return self.r.keys(f"{self.key_prefix}{self._serialize_path(*path)}*")
 
-    def get_string_keys_from_partial(self, *path) -> list:
+    def get_string_keys_from_partial(self, *path: str) -> list[str]:
         keys = self.get_keys_from_partial(*path)
         return self._keys_to_strings(keys, key_prefix=self.key_prefix)
 
-    def get_dict_from_partial(self, *path):
+    def get_dict_from_partial(self, *path: str) -> dict[str, Any]:
         keys = self.get_keys_from_partial(*path)
         root = self._serialize_path(*path) + "/"
         stringified_keys = self._keys_to_strings(keys, self.key_prefix)
-        results = {key[len(root):]: self.cache.get(key) for key in stringified_keys}  # TODO key.removeprefix(root) when py3.9 is released (Oct 20)
+        results = {key.removeprefix(root): self.cache.get(key) for key in stringified_keys}
         tree = self._build_tree(results)
         return tree
 
-    def get_from_cache(self, *path):
+    def get_from_cache(self, *path: str) -> Any:
         """Access a value from the global cache"""
         return self.cache.get(self._serialize_path(*path))
 
-    def set_to_cache(self, *path, value=None):
+    def set_to_cache(self, *path: str, value: Optional[Any] = None) -> Any:
         """Access a value from the global cache"""
         return self.cache.set(self._serialize_path(*path), value)
 
-    def build_tree(self):
+    def build_tree(self) -> dict[str, Any]:
         keys = self.r.keys()
         vals = self.cache.get_many(keys)
         pairs = dict(zip(keys, vals))
         tree = self._build_tree(pairs)
         return tree
 
-    def _build_tree(self, pairs: dict) -> dict:
+    def _build_tree(self, pairs: dict[str, str]) -> dict[str, Any]:
         """Builds nested directory tree from key-val dict with path-like keys"""
-        tree = {}
+        tree: dict[str, Any] = {}
         for key, val in pairs.items():
             self._set_by_path(tree, key.split("/"), val)
         return tree
@@ -87,7 +90,7 @@ def get_key_timestamp(self, key: str) -> datetime.datetime:
             return datetime.datetime.fromtimestamp(0, tz=datetime.timezone.utc)  # If key doesn't exist
         return datetime.datetime.fromtimestamp(int(last_accessed), tz=datetime.timezone.utc)
 
-    def save_to_disk(self):
+    def save_to_disk(self) -> None:
         # Ensure state is not changed whilst getting data
         lock_key = "saving"
         with self.r.lock(lock_key, timeout=5, blocking_timeout=2.5):
@@ -118,7 +121,7 @@ def save_to_disk(self):
         json_pairs = json.dumps(pairs)
         self.cache_path.write_text(json_pairs, encoding="UTF8")
 
-    def load_from_disk(self):
+    def load_from_disk(self) -> None:
         try:
             loaded_json = json.loads(self.cache_path.read_text(encoding="UTF8"))
         except (FileNotFoundError, json.JSONDecodeError):

src/components/homepage.py

@@ -3,7 +3,7 @@
 import dash_core_components as dcc
 import dash_html_components as html
 
-from src.components.navbar import Navbar
+from src.components import navbar
 
 from typing import TYPE_CHECKING
 
@@ -23,8 +23,8 @@
 ], className="page-container static-container vertical-center home")
 
 
-def Homepage(app: Dash):
+def Homepage(app: Dash) -> html.Div:
     return html.Div([
-        Navbar(app),
+        navbar.Navbar(app),
         body
     ], className="page")

src/components/navbar.py

@@ -1,16 +1,17 @@
 from __future__ import annotations
 
+from dash.dependencies import ClientsideFunction, Input, Output, State
 import dash_core_components as dcc
 import dash_html_components as html
-from dash.dependencies import ClientsideFunction, Input, Output, State
 
 from typing import TYPE_CHECKING
 
 if TYPE_CHECKING:
     from dash import Dash
+    from dash.development.base_component import Component
 
 
-def Navbar(app: Dash, download: bool = False):
+def Navbar(app: Dash, download: bool = False) -> html.Nav:
     return html.Nav([
         html.Div([
             dcc.Link([
@@ -22,7 +23,7 @@ def Navbar(app: Dash, download: bool = False):
     ], className="navigation bg-nav")
 
 
-def generate_nav_links(query: str = "", download: bool = False):
+def generate_nav_links(query: str = "", download: bool = False) -> list[Component]:
     query = query if query is not None else ""
     links = []
     if download:
@@ -34,12 +35,12 @@ def generate_nav_links(query: str = "", download: bool = False):
     return links
 
 
-def setup_callbacks(app: Dash):
+def setup_callbacks(app: Dash) -> None:
     @app.callback(Output("nav-list", "children"),
                   [Input('url', 'href')],
                   [State("report-query", "data")],
                   prevent_initial_call=True)
-    def update_nav_links(_, query_string):
+    def update_nav_links(_: str, query_string: str) -> list[Component]:
         return generate_nav_links(query_string)
 
     # app.clientside_callback(