fix: mypy

Signed-off-by: Guilhem Barthes <[email protected]>

Makefile

@@ -32,7 +32,7 @@ format:  ## Format code
 lint:  ## Perform a static analysis of the code
 	flake8 $(SRC_DIRS)
 	bandit --ini=.bandit
-	mypy backend/substrapp/tasks/
+	mypy
 
 .PHONY: shell
 shell:	## Start a Python shell for the Django project

backend/backend/settings/common.py

@@ -210,7 +210,7 @@
 # Used by the Secure aggregation mechanism to retrieve chainkeys
 K8S_SECRET_NAMESPACE = os.getenv("K8S_SECRET_NAMESPACE", "default")
 
-REGISTRY = os.getenv("REGISTRY")
+REGISTRY = os.getenv("REGISTRY", "")
 REGISTRY_SCHEME = os.getenv("REGISTRY_SCHEME")
 REGISTRY_PULL_DOMAIN = os.getenv("REGISTRY_PULL_DOMAIN")
 REGISTRY_IS_LOCAL = to_bool(os.environ.get("REGISTRY_IS_LOCAL"))

backend/builder/exceptions.py

@@ -23,7 +23,7 @@ class BuildRetryError(_ComputeTaskError, CeleryRetryError):
 
     error_type = ComputeTaskErrorType.BUILD_ERROR
 
-    def __init__(self, logs: str, *args, **kwargs):
+    def __init__(self, logs: str, *args: list, **kwargs: dict):
         self.logs = BytesIO(str.encode(logs))
         super().__init__(logs, *args, **kwargs)
 
@@ -37,6 +37,6 @@ class BuildError(_ComputeTaskError, CeleryNoRetryError):
 
     error_type = ComputeTaskErrorType.BUILD_ERROR
 
-    def __init__(self, logs: str, *args, **kwargs):
+    def __init__(self, logs: str, *args: list, **kwargs: dict):
         self.logs = BytesIO(str.encode(logs))
         super().__init__(logs, *args, **kwargs)

backend/builder/image_builder/image_builder.py

@@ -1,6 +1,7 @@
 import json
 import os
 from tempfile import TemporaryDirectory
+from typing import Union
 
 import kubernetes
 import structlog
@@ -174,7 +175,7 @@ def _build_container_image(path: str, tag: str) -> None:
     _delete_kaniko_pod(create_pod, k8s_client, pod_name)
 
 
-def _assert_dockerfile_exist(dockerfile_path: os.PathLike) -> None:
+def _assert_dockerfile_exist(dockerfile_path: Union[str, os.PathLike]) -> None:
     dockerfile_fullpath = os.path.join(dockerfile_path, "Dockerfile")
     if not os.path.exists(dockerfile_fullpath):
         raise BuildError(f"Dockerfile does not exist : {dockerfile_fullpath}")

backend/builder/tasks/task.py

@@ -65,14 +65,14 @@ def on_failure(
     #         {"compute_task_key": compute_task_key, "error_type": error_type, "logs_address": logs_address}
     #     )
 
-    def before_start(self, task_id, args, kwargs):
+    def before_start(self, task_id: str, args: tuple, kwargs: dict) -> None:
         function_key, channel_name = self.get_task_info(args, kwargs)
         with get_orchestrator_client(channel_name) as client:
             client.update_function_status(
                 function_key=function_key, action=orchestrator.function_pb2.FUNCTION_ACTION_BUILDING
             )
 
     def get_task_info(self, args: tuple, kwargs: dict) -> tuple[str, str]:
-        function = orchestrator.Function.parse_raw(kwargs.get("function_serialized"))
-        channel_name = kwargs.get("channel_name")
+        function = orchestrator.Function.parse_raw(kwargs["function_serialized"])
+        channel_name = kwargs["channel_name"]
         return function.key, channel_name

backend/builder/tasks/tasks_build_image.py

@@ -19,7 +19,7 @@
 # Ack late and reject on worker lost allows use to
 # see http://docs.celeryproject.org/en/latest/userguide/configuration.html#task-reject-on-worker-lost
 # and https://github.com/celery/celery/issues/5106
-def build_image(task: BuildTask, function_serialized: str, channel_name: str) -> tuple[str, str]:
+def build_image(task: BuildTask, function_serialized: str, channel_name: str) -> None:
     function = orchestrator.Function.parse_raw(function_serialized)
 
     attempt = 0

backend/builder/volumes.py

@@ -3,5 +3,5 @@
 from django.conf import settings
 
 
-def get_docker_cache_pvc_name():
+def get_docker_cache_pvc_name() -> str:
     return f"{settings.WORKER_PVC_DOCKER_CACHE}-{os.getenv('HOSTNAME')}"

backend/substrapp/tasks/tasks_save_image.py

@@ -52,8 +52,8 @@ def on_failure(
 
     # Returns (function key, channel)
     def get_task_info(self, args: tuple, kwargs: dict) -> tuple[str, str]:
-        function = orchestrator.Function.parse_raw(kwargs.get("function_serialized"))
-        channel_name = kwargs.get("channel_name")
+        function = orchestrator.Function.parse_raw(kwargs["function_serialized"])
+        channel_name = kwargs["channel_name"]
         return function.key, channel_name
 
     def on_success(self, retval: dict[str, Any], task_id: str, args: tuple, kwargs: dict[str, Any]) -> None:

pyproject.toml

@@ -83,6 +83,13 @@ check_untyped_defs = true
 ignore_missing_imports = true
 cache_dir = "/dev/null"
 follow_imports = "silent"
+packages = [
+    "builder",
+    "substrapp.tasks",
+]
+exclude = [
+    "tests/*",
+]
 
 [tool.django-stubs]
 django_settings_module = "backend.settings.test"