chore: add mypy to ci

.mise.toml

@@ -22,11 +22,11 @@ run = ["uv run py.test"]
 run = ["uv run mypy ."]
 
 [tasks."ci"]
-depends = ["lint", "format", "test"]
+depends = ["lint", "format", "test", "mypy"]
 
 [tasks."helm:lint"]
 run = "helm install --dry-run --debug -f charts/gitops/values.yaml debug charts/gitops"
 
 [tasks.release]
 description = "Bump release versions across all files"
-run = "python release.py"
+run = "python release.py"

.pre-commit-config.yaml

@@ -1,12 +1,12 @@
 repos:
 - repo: https://github.com/charliermarsh/ruff-pre-commit
-  rev: v0.2.0
+  rev: v0.6.3
   hooks:
   - id: ruff
     args:
     - --fix
 - repo: https://github.com/charliermarsh/ruff-pre-commit
-  rev: v0.2.0
+  rev: v0.6.3
   hooks:
   - id: ruff-format
 - repo: https://github.com/commitizen-tools/commitizen

README.md

@@ -40,8 +40,6 @@ This is a git repository that you set up, where you list out all of your applica
 
 Secrets should be placed in `secrets.env`. The example file `secrets.example.env` has the environment variables you will need to supply.
 
-Gitops has a helm chart defining its deployment. Invoke scripts are provided to make deployment painless. See `tasks.py`.
-
 Add `export GITOPS_APPS_DIRECTORY=~/<cluster-apps-folder>` to invoke gitops from any directory.
 
 Ensure that gitops has `edit` access to the namespace it is deploying to. An example RoleBinding is:

gitops/__init__.py

@@ -2,7 +2,6 @@
 import sys
 from pathlib import Path
 
-from . import monkeypatches  # NOQA
 from .utils.cli import success, warning
 
 __version__ = "0.11.0"

gitops/common/app.py

@@ -155,7 +155,7 @@ def __init__(self, definition: dict | str):
             # for backwards compat, any chart definition which is a string, is a git repo
             self.type = "git"
             self.git_sha = None
-            self.git_repo_url = definition
+            self.git_repo_url = definition or None
         elif isinstance(definition, dict):
             self.type = definition["type"]
             self.git_sha = definition.get("git_sha")

gitops/core.py

@@ -372,7 +372,7 @@ def git_push(cluster_path: str, retry: int = 3):
     attempts = 0
     while attempts <= retry:
         result = run(f"cd {cluster_path}; git push", warn=True)
-        if result.exited == 1 and "remote contains work" in result.stderr:
+        if result and result.exited == 1 and "remote contains work" in result.stderr:
             attempts += 1
             run(f"cd {cluster_path}; git pull --rebase=true")
         else:

gitops/db.py

@@ -6,7 +6,7 @@
 import time
 
 import boto3
-import dsnparse
+import dsnparse  # type: ignore[import]
 from invoke import run, task
 
 from .utils import kube

gitops/main.py

@@ -13,6 +13,6 @@
         namespace.add_task(task)
 
 # Namespace the rarer ones.
-namespace.add_collection(db)
+namespace.add_collection(db)  # type: ignore
 
 program = Program(namespace=namespace, version=version)

gitops/utils/__init__.py

@@ -13,13 +13,17 @@ def gen_secret(length: int = 64) -> str:
 
 
 def get_account_id() -> str:
+    # TODO REMOVE THIS FUNCTION and account_id in general
+
     if "ACCOUNT_ID" not in CACHE:
         # This is not ideal, as it makes an assumption that the account_id of interest is the
         # one the user is currently sitting in. Ideally, should ask the cluster (though that is
         # messy in its own right, since we don't necessarily want the cluster that's in the
         # current context either).
-        caller_identity = run("aws sts get-caller-identity", hide=True).stdout.strip()
-        CACHE["ACCOUNT_ID"] = json.loads(caller_identity)["Account"]
+        result = run("aws sts get-caller-identity", hide=True)
+        if result:
+            caller_identity = result.stdout.strip()
+            CACHE["ACCOUNT_ID"] = json.loads(caller_identity)["Account"]
     return CACHE["ACCOUNT_ID"]
 
 

gitops/utils/apps.py

@@ -1,5 +1,5 @@
 import sys
-from pathlib import PosixPath
+from pathlib import Path
 
 from colorama import Fore
 from tabulate import tabulate
@@ -15,7 +15,7 @@
 from .tags import colour_tags, validate_tags
 
 
-def is_valid_app_directory(directory: PosixPath) -> bool:
+def is_valid_app_directory(directory: Path) -> bool:
     files = ["deployment.yml", "secrets.yml"]
     file_paths = [(directory / file).is_file() for file in files]
     return all(file_paths)

gitops/utils/cli.py

@@ -1,4 +1,5 @@
 from collections.abc import Callable
+
 from colorama import Fore
 
 

gitops/utils/images.py

@@ -58,7 +58,7 @@ def colour_image(image_tag: str) -> str:
         return colourise(bits[0], color_hash(bits[0]))
 
 
-def color_hash(bit: str) -> Fore:  # type: ignore[no-untyped-def]
+def color_hash(bit: str) -> str:
     return [
         Fore.RED,
         Fore.GREEN,

gitops/utils/kube.py

@@ -20,7 +20,7 @@
 import yaml
 from colorama import Fore
 from invoke import run
-from invoke.exceptions import UnexpectedExit
+from invoke.exceptions import Failure, UnexpectedExit
 
 from gitops.common.app import App
 from gitops.settings import get_apps_directory
@@ -55,8 +55,8 @@ async def run_job(
         "image": app.image,
         "serviceAccountName": app.service_account_name,
     }
-    extra_labels = {}
-    container_resources = {}
+    extra_labels: dict[str, str] = {}
+    container_resources: ContainerResources | None = None
 
     if cpu and memory:
         cpu_request = f"{cpu}m" if cpu else ""
@@ -275,10 +275,10 @@ async def _find_pod() -> str:
                         _, _, output_log = await async_run(cmd)
                         logs += output_log
 
-                except Exception as e:
+                except Failure as e:
                     if "current phase is Succeeded" not in str(e.result.stdout):
                         raise e
-        except Exception as e:
+        except Failure as e:
             if "current phase is Succeeded" not in str(e.result.stdout):
                 raise e
         finally:
@@ -292,9 +292,9 @@ async def wait_for_pod(context: str, namespace: str, pod: str) -> str | None:
     while True:
         cmd = "kubectl get pod" f" --context {context}" f" -n {namespace}" ' -o jsonpath="{.status.phase}"' f" {pod}"
         stdout, _, _ = await async_run(cmd)
-        stdout = stdout.decode().lower()
-        if stdout != "pending":
-            return stdout
+        output = stdout.decode().lower()
+        if output != "pending":
+            return output
 
 
 def retry(*args: object, max_attempts: int = 3, delay: int = 1):  # noqa: C901

gitops/utils/yaml.py

@@ -24,12 +24,12 @@ def increase_indent(self, flow=False, indentless=False):
 if pyyaml.safe_dump is pyyaml.dump:
     # PyYAML v4.1
     SafeDumper = pyyaml.dumper.Dumper
-    DangerDumper = pyyaml.dumper.DangerDumper
+    DangerDumper = pyyaml.dumper.DangerDumper  # type: ignore
 else:
-    SafeDumper = pyyaml.dumper.SafeDumper
+    SafeDumper = pyyaml.dumper.SafeDumper  # type: ignore
     DangerDumper = pyyaml.dumper.Dumper
 
-SafeDumper.increase_indent = increase_indent
+SafeDumper.increase_indent = increase_indent  # type: ignore
 DangerDumper.increase_indent = increase_indent
 
 pyyaml.add_representer(dict, map_representer, Dumper=SafeDumper)
@@ -43,4 +43,4 @@ def increase_indent(self, flow=False, indentless=False):
 # Merge PyYAML namespace into ours.
 # This allows users a drop-in replacement:
 #   import utils.yaml as yaml
-from yaml import *  # noqa isort:skip
+from yaml import *  # type: ignore # noqa isort:skip

gitops_server/main.py

@@ -7,7 +7,6 @@
 
 from gitops_server import settings
 from gitops_server.app import app
-from gitops_server.logging_config import *  # noqa
 from gitops_server.workers import DeploymentStatusWorker, DeployQueueWorker
 
 logging.basicConfig(level=logging.INFO)

gitops_server/utils/__init__.py

@@ -19,7 +19,7 @@ async def run(command, suppress_errors=False) -> RunOutput:
     )
 
     stdout, stderr = await proc.communicate()
-    exit_code = proc.returncode
+    exit_code = proc.returncode or 1
     if exit_code == 0:
         return RunOutput(exit_code=exit_code, output=stdout.decode())
     else:

gitops_server/utils/github.py

@@ -8,8 +8,7 @@
 GITHUB_OAUTH_TOKEN = os.environ.get("GITHUB_OAUTH_TOKEN")
 
 
-class IssueNotFound(Exception):
-    ...
+class IssueNotFound(Exception): ...
 
 
 class STATUSES:

gitops_server/utils/slack.py

@@ -37,10 +37,11 @@ async def post(message: str) -> None:
     logger.info("POSTING TO SLACK")
     data = {"text": message}
     async with httpx.AsyncClient() as client:
-        response = await client.post(SLACK_URL, json=data)
-        if response.status_code >= 300:
-            logger.warning("Failed to post a message to slack (see below):")
-            logger.error(f"{message}", exc_info=True)
+        if SLACK_URL:
+            response = await client.post(SLACK_URL, json=data)
+            if response.status_code >= 300:
+                logger.warning("Failed to post a message to slack (see below):")
+                logger.error(f"{message}", exc_info=True)
 
 
 async def find_commiter_slack_user(name: str, email: str) -> SlackUser | None:

gitops_server/workers/deployer/deploy.py

@@ -33,10 +33,9 @@ async def post_init_summary(source, username, added_apps, updated_apps, removed_
 
 async def post_result(app: App, result: UpdateAppResult, deployer: "Deployer", **kwargs):
     if result["exit_code"] != 0:
-        result = await handle_failed_deploy(app, result, deployer)
-
+        deploy_result = await handle_failed_deploy(app, result, deployer)
         message = (
-            result.get("slack_message")
+            deploy_result["slack_message"]
             or f"Failed to deploy app `{result['app_name']}` for cluster `{settings.CLUSTER_NAME}`:\n>>>{result['output']}"
         )
 
@@ -139,7 +138,8 @@ async def uninstall_app(self, app: App) -> UpdateAppResult:
         async with self.semaphore:
             logger.info(f"Uninstalling app {app.name!r}.")
             result = await run(f"helm uninstall {app.name} -n {app.namespace}", suppress_errors=True)
-            update_result = UpdateAppResult(app_name=app.name, **result)
+            if result:
+                update_result = UpdateAppResult(app_name=app.name, slack_message="", **result)
             await post_result(
                 app=app,
                 result=update_result,
@@ -156,6 +156,7 @@ async def update_app_deployment(self, app: App) -> UpdateAppResult | None:
         async with self.semaphore:
             logger.info(f"Deploying app {app.name!r}.")
             if app.chart.type == "git":
+                assert app.chart.git_repo_url
                 async with temp_repo(app.chart.git_repo_url, sha=app.chart.git_sha) as chart_folder_path:
                     await run(f"cd {chart_folder_path}; helm dependency build")
                     with tempfile.NamedTemporaryFile(suffix=".yml") as cfg:
@@ -195,7 +196,7 @@ async def update_app_deployment(self, app: App) -> UpdateAppResult | None:
                 logger.warning("Local is not implemented yet")
                 return None
 
-            update_result = UpdateAppResult(app_name=app.name, **result)
+            update_result = UpdateAppResult(app_name=app.name, slack_message="", **result)
 
             await post_result(app=app, result=update_result, deployer=self)
         return update_result

gitops_server/workers/deployer/hooks.py

@@ -9,7 +9,7 @@
 from gitops_server import settings
 from gitops_server.types import UpdateAppResult
 from gitops_server.utils import github
-from gitops_server.utils.slack import SlackGroup, find_commiter_slack_user
+from gitops_server.utils.slack import SlackGroup, SlackUser, find_commiter_slack_user
 
 logger = logging.getLogger(__name__)
 
@@ -74,7 +74,7 @@ async def handle_failed_deploy(app: App, result: UpdateAppResult, deployer) -> U
     email = deployer.author_email
 
     if "devops" in email.lower() or "tickforge" in email.lower():
-        slack_user = DEFAULT_USER_GROUP
+        slack_user: SlackGroup | SlackUser = DEFAULT_USER_GROUP
     else:
         slack_user = (
             await find_commiter_slack_user(name=deployer.author_name, email=deployer.author_email) or DEFAULT_USER_GROUP

gitops_server/workers/status_updater/worker.py

@@ -14,9 +14,9 @@
 import asyncio
 import logging
 
-import kubernetes_asyncio
-import kubernetes_asyncio.client
-import kubernetes_asyncio.config
+import kubernetes_asyncio  # type:ignore[import-untyped]
+import kubernetes_asyncio.client  # type:ignore[import-untyped]
+import kubernetes_asyncio.config  # type:ignore[import-untyped]
 
 from gitops_server.settings import CLUSTER_NAMESPACE
 from gitops_server.utils import github

pyproject.toml

@@ -37,6 +37,11 @@ build-backend = "hatchling.build"
 dev-dependencies = [
     "pytest>=8.3.2",
     "ruff>=0.6.3",
+    "types-colorama>=0.4.15.20240311",
+    "mypy>=1.11.2",
+    "types-pyyaml>=6.0.12.20240808",
+    "types-tabulate>=0.9.0.20240106",
+    "boto3-stubs>=1.35.11",
 ]
 
 
@@ -50,3 +55,6 @@ style = [
     'py://nitpick/resources/any/commitizen',
 ]
 cache = "never"
+
+[tool.mypy]
+exclude = ["tests"]