Typechecking for manifests (opensearch-project#1224)

* Adding python type checking for src/manifests

Signed-off-by: Peter Nied <[email protected]>

.github/workflows/python-tests.yml

@@ -31,7 +31,7 @@ jobs:
           pipenv run flake8 .
       - name: Run Type Checker
         run: |
-          pipenv run mypy .
+          pipenv run mypy src/manifests tests/tests_manifests
       - name: Run Tests with Coverage
         run: |
           pipenv run coverage run -m pytest --cov=./src --cov-report=xml

.pre-commit-config.yaml

@@ -18,7 +18,7 @@ repos:
   - id: mypy
     stages: [commit]
     name: mypy
-    entry: bash -c 'pipenv run mypy .'
+    entry: bash -c 'pipenv run mypy src/manifests tests/tests_manifests'
     language: system
   - id: pytest
     stages: [commit]

pyproject.toml

@@ -4,4 +4,13 @@ target-version = ['py37']
 
 [tool.isort]
 profile = "black"
-line_length = 160
+line_length = 160
+
+[tool.mypy]
+python_version = 3.7
+disallow_untyped_defs = true
+warn_return_any = true
+warn_unused_configs = true
+strict_optional = false
+show_error_codes = true
+ignore_missing_imports = true

src/build_workflow/builders.py

@@ -8,15 +8,15 @@
 
 from build_workflow.builder_from_dist import BuilderFromDist
 from build_workflow.builder_from_source import BuilderFromSource
-from manifests.input_manifest import InputManifest
+from manifests.input_manifest import InputComponentFromDist, InputComponentFromSource
 
 
 class Builders(ABC):
     @classmethod
     def builder_from(self, component, target):
-        if type(component) is InputManifest.ComponentFromDist:
+        if type(component) is InputComponentFromDist:
             return BuilderFromDist(component, target)
-        elif type(component) is InputManifest.ComponentFromSource:
+        elif type(component) is InputComponentFromSource:
             return BuilderFromSource(component, target)
         else:
             raise ValueError(f"Invalid component type: {type(component)}")

src/ci_workflow/ci_check_lists.py

@@ -8,15 +8,15 @@
 
 from ci_workflow.ci_check_list_dist import CiCheckListDist
 from ci_workflow.ci_check_list_source import CiCheckListSource
-from manifests.input_manifest import InputManifest
+from manifests.input_manifest import InputComponentFromDist, InputComponentFromSource
 
 
 class CiCheckLists(ABC):
     @classmethod
     def from_component(self, component, target):
-        if type(component) is InputManifest.ComponentFromDist:
+        if type(component) is InputComponentFromDist:
             return CiCheckListDist(component, target)
-        elif type(component) is InputManifest.ComponentFromSource:
+        elif type(component) is InputComponentFromSource:
             return CiCheckListSource(component, target)
         else:
             raise ValueError(f"Invalid component type: {type(component)}")

src/git/git_repository.py

@@ -8,18 +8,20 @@
 import os
 import subprocess
 from pathlib import Path
+from typing import Any, List
 
 from system.temporary_directory import TemporaryDirectory
 
 
 class GitRepository:
+    dir: str
     """
     This class checks out a Git repository at a particular ref into an empty named directory (or temporary a directory if no named directory is given).
     Temporary directories will be automatically deleted when the GitRepository object goes out of scope; named directories will be left alone.
     Clients can obtain the actual commit ID by querying the "sha" attribute, and the temp directory name with "dir".
     """
 
-    def __init__(self, url, ref, directory=None, working_subdirectory=None):
+    def __init__(self, url: str, ref: str, directory: str = None, working_subdirectory: str = None) -> None:
         self.url = url
         self.ref = ref
         if directory is None:
@@ -32,15 +34,14 @@ def __init__(self, url, ref, directory=None, working_subdirectory=None):
         self.working_subdirectory = working_subdirectory
         self.__checkout__()
 
-    def __enter__(self):
+    def __enter__(self) -> 'GitRepository':
         return self
 
-    def __exit__(self, exc_type, exc_value, exc_traceback):
+    def __exit__(self, exc_type: Any, exc_value: Any, exc_traceback: Any) -> None:
         if self.temp_dir:
             self.temp_dir.__exit__(exc_type, exc_value, exc_traceback)
 
-    def __checkout__(self):
-        # Check out the repository
+    def __checkout__(self) -> None:
         self.execute_silent("git init", self.dir)
         self.execute_silent(f"git remote add origin {self.url}", self.dir)
         self.execute_silent(f"git fetch --depth 1 origin {self.ref}", self.dir)
@@ -49,18 +50,18 @@ def __checkout__(self):
         logging.info(f"Checked out {self.url}@{self.ref} into {self.dir} at {self.sha}")
 
     @property
-    def working_directory(self):
+    def working_directory(self) -> str:
         if self.working_subdirectory:
             return os.path.join(self.dir, self.working_subdirectory)
         else:
             return self.dir
 
     @classmethod
-    def stable_ref(self, url, ref):
+    def stable_ref(self, url: str, ref: str) -> List[str]:
         results = subprocess.check_output(f"git ls-remote {url} {ref}", shell=True).decode().strip().split("\t")
         return results if len(results) > 1 else [ref, ref]
 
-    def execute_silent(self, command, cwd=None):
+    def execute_silent(self, command: str, cwd: str = None) -> None:
         cwd = cwd or self.working_directory
         logging.info(f'Executing "{command}" in {cwd}')
         subprocess.check_call(
@@ -71,17 +72,17 @@ def execute_silent(self, command, cwd=None):
             stderr=subprocess.DEVNULL,
         )
 
-    def output(self, command, cwd=None):
+    def output(self, command: str, cwd: str = None) -> str:
         cwd = cwd or self.working_directory
         logging.info(f'Executing "{command}" in {cwd}')
         return subprocess.check_output(command, cwd=cwd, shell=True).decode().strip()
 
-    def execute(self, command, cwd=None):
+    def execute(self, command: str, cwd: str = None) -> None:
         cwd = cwd or self.working_directory
         logging.info(f'Executing "{command}" in {cwd}')
         subprocess.check_call(command, cwd=cwd, shell=True)
 
-    def path(self, subdirname=None):
+    def path(self, subdirname: str = None) -> Path:
         dirname = self.dir
         if subdirname:
             dirname = os.path.join(self.dir, subdirname)

src/manifests/build/build_manifest_1_0.py

@@ -4,7 +4,8 @@
 # this file be licensed under the Apache-2.0 license or a
 # compatible open source license.
 
-from manifests.component_manifest import ComponentManifest
+from typing import Any
+from manifests.component_manifest import ComponentManifest, Components, Component
 
 """
 A BuildManifest is an immutable view of the outputs from a build step
@@ -35,7 +36,7 @@
 """
 
 
-class BuildManifest_1_0(ComponentManifest):
+class BuildManifest_1_0(ComponentManifest['BuildManifest_1_0', 'BuildComponents_1_0']):
     SCHEMA = {
         "build": {
             "required": True,
@@ -73,51 +74,54 @@ class BuildManifest_1_0(ComponentManifest):
         },
     }
 
-    def __init__(self, data):
+    def __init__(self, data: Any):
         super().__init__(data)
         self.build = self.Build(data["build"])
 
-    def __to_dict__(self):
+    def __to_dict__(self) -> dict:
         return {
             "schema-version": "1.0",
             "build": self.build.__to_dict__(),
             "components": self.components.__to_dict__()
         }
 
-    class Components(ComponentManifest.Components):
-        def __create__(self, data):
-            return BuildManifest_1_0.Component(data)
-
     class Build:
-        def __init__(self, data):
+        def __init__(self, data: Any):
             self.name = data["name"]
             self.version = data["version"]
             self.architecture = data["architecture"]
             self.id = data["id"]
 
-        def __to_dict__(self):
+        def __to_dict__(self) -> dict:
             return {
                 "name": self.name,
                 "version": self.version,
                 "architecture": self.architecture,
                 "id": self.id
             }
 
-    class Component(ComponentManifest.Component):
-        def __init__(self, data):
-            super().__init__(data)
-            self.repository = data["repository"]
-            self.ref = data["ref"]
-            self.commit_id = data["commit_id"]
-            self.artifacts = data.get("artifacts", {})
-            self.version = data["version"]
 
-        def __to_dict__(self):
-            return {
-                "name": self.name,
-                "repository": self.repository,
-                "ref": self.ref,
-                "commit_id": self.commit_id,
-                "artifacts": self.artifacts,
-                "version": self.version,
-            }
+class BuildComponents_1_0(Components['BuildComponent_1_0']):
+    @classmethod
+    def __create__(self, data: Any) -> 'BuildComponent_1_0':
+        return BuildComponent_1_0(data)
+
+
+class BuildComponent_1_0(Component):
+    def __init__(self, data: Any):
+        super().__init__(data)
+        self.repository = data["repository"]
+        self.ref = data["ref"]
+        self.commit_id = data["commit_id"]
+        self.artifacts = data.get("artifacts", {})
+        self.version = data["version"]
+
+    def __to_dict__(self) -> dict:
+        return {
+            "name": self.name,
+            "repository": self.repository,
+            "ref": self.ref,
+            "commit_id": self.commit_id,
+            "artifacts": self.artifacts,
+            "version": self.version,
+        }

src/manifests/build/build_manifest_1_1.py

@@ -4,7 +4,8 @@
 # this file be licensed under the Apache-2.0 license or a
 # compatible open source license.
 
-from manifests.component_manifest import ComponentManifest
+from typing import Any
+from manifests.component_manifest import ComponentManifest, Components, Component
 
 """
 A BuildManifest is an immutable view of the outputs from a build step
@@ -36,7 +37,7 @@
 """
 
 
-class BuildManifest_1_1(ComponentManifest):
+class BuildManifest_1_1(ComponentManifest['BuildManifest_1_1', 'BuildComponents_1_1']):
     SCHEMA = {
         "build": {
             "required": True,
@@ -74,51 +75,54 @@ class BuildManifest_1_1(ComponentManifest):
         },
     }
 
-    def __init__(self, data):
+    def __init__(self, data: Any):
         super().__init__(data)
         self.build = self.Build(data["build"])
 
-    def __to_dict__(self):
+    def __to_dict__(self) -> dict:
         return {
             "schema-version": "1.1",
             "build": self.build.__to_dict__(),
             "components": self.components.__to_dict__()
         }
 
     class Build:
-        def __init__(self, data):
+        def __init__(self, data: Any):
             self.name = data["name"]
             self.version = data["version"]
             self.architecture = data["architecture"]
             self.id = data["id"]
 
-        def __to_dict__(self):
+        def __to_dict__(self) -> dict:
             return {
                 "name": self.name,
                 "version": self.version,
                 "architecture": self.architecture,
                 "id": self.id
             }
 
-    class Components(ComponentManifest.Components):
-        def __create__(self, data):
-            return BuildManifest_1_1.Component(data)
 
-    class Component(ComponentManifest.Component):
-        def __init__(self, data):
-            super().__init__(data)
-            self.repository = data["repository"]
-            self.ref = data["ref"]
-            self.commit_id = data["commit_id"]
-            self.artifacts = data.get("artifacts", {})
-            self.version = data["version"]
+class BuildComponents_1_1(Components['BuildComponent_1_1']):
+    @classmethod
+    def __create__(self, data: Any) -> 'BuildComponent_1_1':
+        return BuildComponent_1_1(data)
 
-        def __to_dict__(self):
-            return {
-                "name": self.name,
-                "repository": self.repository,
-                "ref": self.ref,
-                "commit_id": self.commit_id,
-                "artifacts": self.artifacts,
-                "version": self.version,
-            }
+
+class BuildComponent_1_1(Component):
+    def __init__(self, data: Any):
+        super().__init__(data)
+        self.repository = data["repository"]
+        self.ref = data["ref"]
+        self.commit_id = data["commit_id"]
+        self.artifacts = data.get("artifacts", {})
+        self.version = data["version"]
+
+    def __to_dict__(self) -> dict:
+        return {
+            "name": self.name,
+            "repository": self.repository,
+            "ref": self.ref,
+            "commit_id": self.commit_id,
+            "artifacts": self.artifacts,
+            "version": self.version,
+        }