Added type hinting (equinor#195)

Author: asnyv

.gitignore

@@ -1,5 +1,7 @@
 .eggs
 *.egg-info
+*~
+.mypy_cache
 __pycache__
 node_modules
 venv

.travis.yml

@@ -35,6 +35,7 @@ script:
   - black --check webviz_config tests setup.py
   - pylint webviz_config tests setup.py
   - bandit -r -c ./bandit.yml webviz_config tests setup.py
+  - mypy --package webviz_config --ignore-missing-imports --disallow-untyped-defs --show-error-codes
 
   - webviz certificate
   - pytest tests --forked

setup.py

@@ -10,6 +10,7 @@
     "pytest-xdist",
     "black",
     "bandit",
+    "mypy",
 ]
 
 setup(

webviz_config/_build_webviz.py

@@ -3,6 +3,7 @@
 import shutil
 import tempfile
 import subprocess  # nosec
+import argparse
 
 from yaml import YAMLError
 
@@ -15,7 +16,7 @@
 STATIC_FOLDER = os.path.join(os.path.dirname(__file__), "static")
 
 
-def build_webviz(args):
+def build_webviz(args: argparse.Namespace) -> None:
 
     if args.theme not in installed_themes:
         raise ValueError(f"Theme `{args.theme}` is not installed.")
@@ -87,7 +88,7 @@ def build_webviz(args):
             shutil.rmtree(build_directory)
 
 
-def run_webviz(args, build_directory):
+def run_webviz(args: argparse.Namespace, build_directory: str) -> None:
 
     print(
         f"{terminal_colors.YELLOW}"

webviz_config/_certificate.py

@@ -5,6 +5,7 @@
 import getpass
 import datetime
 import subprocess  # nosec
+import argparse
 
 from cryptography import x509
 from cryptography.x509.oid import NameOID
@@ -32,7 +33,7 @@
 SERVER_CRT_FILENAME = "server.crt"
 
 
-def user_data_dir():
+def user_data_dir() -> str:
     """Returns platform specific path to store user application data
     """
 
@@ -45,7 +46,7 @@ def user_data_dir():
     return os.path.expanduser("~/.local/share/webviz")
 
 
-def create_key(key_path):
+def create_key(key_path: str) -> rsa.RSAPrivateKey:
 
     key = rsa.generate_private_key(
         public_exponent=65537, key_size=2048, backend=default_backend()
@@ -63,7 +64,12 @@ def create_key(key_path):
     return key
 
 
-def certificate_template(subject, issuer, public_key, certauthority=False):
+def certificate_template(
+    subject: x509.name.Name,
+    issuer: x509.name.Name,
+    public_key: x509.name.Name,
+    certauthority: bool = False,
+) -> x509.base.CertificateBuilder:
 
     if certauthority:
         not_valid_after = datetime.datetime.utcnow() + datetime.timedelta(days=365 * 10)
@@ -88,7 +94,7 @@ def certificate_template(subject, issuer, public_key, certauthority=False):
     )
 
 
-def create_ca(args):
+def create_ca(args: argparse.Namespace) -> None:
 
     directory = user_data_dir()
 
@@ -180,8 +186,7 @@ def create_ca(args):
         )
 
 
-def create_certificate(directory):
-
+def create_certificate(directory: str) -> None:
     ca_directory = user_data_dir()
     ca_key_path = os.path.join(ca_directory, CA_KEY_FILENAME)
     ca_crt_path = os.path.join(ca_directory, CA_CRT_FILENAME)

webviz_config/_config_parser.py

@@ -4,6 +4,7 @@
 import inspect
 import importlib
 import typing
+import types
 import warnings
 
 import yaml
@@ -16,26 +17,26 @@
 SPECIAL_ARGS = ["self", "app", "container_settings", "_call_signature", "_imports"]
 
 
-def _get_webviz_plugins(module):
+def _get_webviz_plugins(module: types.ModuleType) -> list:
     """Returns a list of all Webviz plugins
     in the module given as input.
     """
 
-    def _is_webviz_plugin(obj):
+    def _is_webviz_plugin(obj: typing.Any) -> bool:
         return inspect.isclass(obj) and issubclass(obj, WebvizPluginABC)
 
     return [member[0] for member in inspect.getmembers(module, _is_webviz_plugin)]
 
 
 def _call_signature(
-    module,
-    module_name,
-    plugin_name,
-    shared_settings,
-    kwargs,
-    config_folder,
-    contact_person=None,
-):
+    module: types.ModuleType,
+    module_name: str,
+    plugin_name: str,
+    shared_settings: dict,
+    kwargs: dict,
+    config_folder: pathlib.Path,
+    contact_person: typing.Optional[dict] = None,
+) -> tuple:
     # pylint: disable=too-many-branches,too-many-statements
     """Takes as input the name of a plugin, the module it is located in,
     together with user given arguments (originating from the configuration
@@ -158,7 +159,7 @@ class ConfigParser:
 
     STANDARD_PLUGINS = _get_webviz_plugins(standard_plugins)
 
-    def __init__(self, yaml_file):
+    def __init__(self, yaml_file: str):
 
         ConfigParser.check_for_tabs_in_file(yaml_file)
 
@@ -181,12 +182,12 @@ def __init__(self, yaml_file):
             ).with_traceback(sys.exc_info()[2])
 
         self._config_folder = pathlib.Path(yaml_file).parent
-        self._page_ids = []
-        self._assets = set()
+        self._page_ids: typing.List[str] = []
+        self._assets: set = set()
         self.clean_configuration()
 
     @staticmethod
-    def check_for_tabs_in_file(path):
+    def check_for_tabs_in_file(path: str) -> None:
 
         with open(path, "r") as filehandle:
             # Create a list with unique entries of line numbers containing tabs
@@ -209,7 +210,7 @@ def check_for_tabs_in_file(path):
                 f"{terminal_colors.END}"
             )
 
-    def _generate_page_id(self, title):
+    def _generate_page_id(self, title: str) -> str:
         """From the user given title, this function provides a unique
         human readable page id, not already present in self._page_ids
         """
@@ -225,7 +226,7 @@ def _generate_page_id(self, title):
 
         return page_id
 
-    def clean_configuration(self):
+    def clean_configuration(self) -> None:
         # pylint: disable=too-many-branches,too-many-statements
         """Various cleaning and checks of the raw configuration read
         from the user provided yaml configuration file.
@@ -372,13 +373,13 @@ def clean_configuration(self):
                     self.assets.update(getattr(module, plugin_name).ASSETS)
 
     @property
-    def configuration(self):
+    def configuration(self) -> dict:
         return self._configuration
 
     @property
-    def shared_settings(self):
+    def shared_settings(self) -> dict:
         return self._shared_settings
 
     @property
-    def assets(self):
+    def assets(self) -> set:
         return self._assets

webviz_config/_is_reload_process.py

@@ -1,7 +1,7 @@
 import os
 
 
-def is_reload_process():
+def is_reload_process() -> bool:
     """Within the flask reload machinery, it is not straight forward to know
     if the code is run as the main process (i.e. the process the user directly
     started), or if the code is a "hot reload process" (see Flask

webviz_config/_localhost_certificate.py

@@ -13,21 +13,21 @@ class LocalhostCertificate:
     only readable by the user running the process, and are deleted on exit.
     """
 
-    def __init__(self):
+    def __init__(self) -> None:
         if not is_reload_process():
             self._ssl_temp_dir = os.environ["WEBVIZ_SSL_TEMP_DIR"] = tempfile.mkdtemp()
             create_certificate(self._ssl_temp_dir)
             atexit.register(self._delete_temp_dir)
         else:
             self._ssl_temp_dir = os.environ["WEBVIZ_SSL_TEMP_DIR"]
 
-    def _delete_temp_dir(self):
+    def _delete_temp_dir(self) -> None:
         """Delete temporary directory with on-the-fly generated localhost certificates
         """
         shutil.rmtree(self._ssl_temp_dir)
 
     @property
-    def ssl_context(self):
+    def ssl_context(self) -> tuple:
         return (
             os.path.join(self._ssl_temp_dir, SERVER_CRT_FILENAME),
             os.path.join(self._ssl_temp_dir, SERVER_KEY_FILENAME),

webviz_config/_localhost_open_browser.py

@@ -11,15 +11,15 @@
 class LocalhostOpenBrowser:
     # pylint: disable=too-few-public-methods
 
-    def __init__(self, port, token):
+    def __init__(self, port: int, token: str):
         self._port = port
         self._token = token
 
         if not is_reload_process():
             # Only open new browser tab if not a reload process
             threading.Thread(target=self._timer).start()
 
-    def _timer(self):
+    def _timer(self) -> None:
         """Waits until the app is ready, and then opens the page
         in the default browser.
         """
@@ -39,14 +39,14 @@ def _timer(self):
             f"{self._url(with_token=True)}"
         )
 
-    def _url(self, with_token=False, https=True):
+    def _url(self, with_token: bool = False, https: bool = True) -> str:
         return (
             f"{'https' if https else 'http'}://localhost:{self._port}"
             + f"{'?ott=' + self._token if with_token else ''}"
         )
 
     @staticmethod
-    def _get_browser_controller():
+    def _get_browser_controller() -> webbrowser.BaseBrowser:
         for browser in ["chrome", "chromium-browser"]:
             try:
                 return webbrowser.get(using=browser)
@@ -57,7 +57,7 @@ def _get_browser_controller():
         # preferred browsers are installed:
         return webbrowser.get()
 
-    def _app_ready(self):
+    def _app_ready(self) -> bool:
         """Check if the flask instance is ready.
         """
 
@@ -67,7 +67,7 @@ def _app_ready(self):
         try:
             urllib.request.urlopen(self._url(https=False))  # nosec
             app_ready = True
-        except urllib.error.URLError:
+        except urllib.error.URLError:  # type: ignore[attr-defined]
             # The flask instance has not started
             app_ready = False
         except ConnectionResetError:
@@ -79,7 +79,7 @@ def _app_ready(self):
 
         return app_ready
 
-    def _open_new_tab(self):
+    def _open_new_tab(self) -> None:
         """Open the url (with token) in the default browser.
         """
 

webviz_config/_localhost_token.py

@@ -30,7 +30,7 @@ class LocalhostToken:
     two different localhost applications running simultaneously do not interfere.
     """
 
-    def __init__(self, app, port):
+    def __init__(self, app: flask.app.Flask, port: int):
         self._app = app
         self._port = port
 
@@ -53,17 +53,17 @@ def __init__(self, app, port):
         self.set_request_decorators()
 
     @staticmethod
-    def generate_token():
+    def generate_token() -> str:
         return secrets.token_urlsafe(nbytes=64)
 
     @property
-    def one_time_token(self):
+    def one_time_token(self) -> str:
         return self._ott
 
-    def set_request_decorators(self):
+    def set_request_decorators(self) -> None:
         # pylint: disable=inconsistent-return-statements
         @self._app.before_request
-        def _check_for_ott_or_cookie():
+        def _check_for_ott_or_cookie():  # type: ignore[no-untyped-def]
             if not self._ott_validated and self._ott == flask.request.args.get("ott"):
                 self._ott_validated = True
                 flask.g.set_cookie_token = True
@@ -77,7 +77,9 @@ def _check_for_ott_or_cookie():
                 flask.abort(401)
 
         @self._app.after_request
-        def _set_cookie_token_in_response(response):
+        def _set_cookie_token_in_response(
+            response: flask.wrappers.Response,
+        ) -> flask.wrappers.Response:
             if "set_cookie_token" in flask.g and flask.g.set_cookie_token:
                 response.set_cookie(
                     key=f"cookie_token_{self._port}", value=self._cookie_token