Type hints in worker_wrapper (#778)

docker-app/worker_wrapper/wrapper.py

@@ -7,7 +7,7 @@
 import uuid
 from datetime import timedelta
 from pathlib import Path
-from typing import Any, Dict, Iterable, List, Tuple
+from typing import Any, Iterable
 
 import docker
 import requests
@@ -45,6 +45,7 @@
 DOCKER_SIGKILL_EXIT_CODE = 137
 QGIS_CONTAINER_NAME = os.environ.get("QGIS_CONTAINER_NAME", None)
 QFIELDCLOUD_HOST = os.environ.get("QFIELDCLOUD_HOST", None)
+TMP_FILE = Path("/tmp")
 
 assert QGIS_CONTAINER_NAME
 assert QFIELDCLOUD_HOST
@@ -63,9 +64,9 @@ def __init__(self, job_id: str) -> None:
         try:
             self.job_id = job_id
             self.job = self.job_class.objects.select_related().get(id=job_id)
-            self.shared_tempdir = Path(tempfile.mkdtemp(dir="/tmp"))
+            self.shared_tempdir = Path(tempfile.mkdtemp(dir=TMP_FILE))
         except Exception as err:
-            feedback: Dict[str, Any] = {}
+            feedback: dict[str, Any] = {}
             (_type, _value, tb) = sys.exc_info()
             feedback["error"] = str(err)
             feedback["error_origin"] = "worker_wrapper"
@@ -82,7 +83,7 @@ def __init__(self, job_id: str) -> None:
             else:
                 logger.critical(msg, exc_info=err)
 
-    def get_context(self) -> Dict[str, Any]:
+    def get_context(self) -> dict[str, Any]:
         context = model_to_dict(self.job)
 
         for key, value in model_to_dict(self.job.project).items():
@@ -92,10 +93,9 @@ def get_context(self) -> Dict[str, Any]:
 
         return context
 
-    def get_command(self) -> List[str]:
-        return [
-            p % self.get_context() for p in ["python3", "entrypoint.py", *self.command]
-        ]
+    def get_command(self) -> list[str]:
+        context = self.get_context()
+        return [p % context for p in ["python3", "entrypoint.py", *self.command]]
 
     def before_docker_run(self) -> None:
         pass
@@ -231,8 +231,8 @@ def run(self):
                 )
 
     def _run_docker(
-        self, command: List[str], volumes: List[str], run_opts: Dict[str, Any] = {}
-    ) -> Tuple[int, bytes]:
+        self, command: list[str], volumes: list[str], run_opts: dict[str, Any] = {}
+    ) -> tuple[int, bytes]:
         QGIS_CONTAINER_NAME = os.environ.get("QGIS_CONTAINER_NAME", None)
         QFIELDCLOUD_HOST = os.environ.get("QFIELDCLOUD_HOST", None)
         QFIELDCLOUD_WORKER_QFIELDCLOUD_URL = os.environ.get(
@@ -421,7 +421,7 @@ def __init__(self, job_id: str) -> None:
         if self.job.overwrite_conflicts:
             self.command = [*self.command, "--overwrite-conflicts"]
 
-    def _prepare_deltas(self, deltas: Iterable[Delta]):
+    def _prepare_deltas(self, deltas: Iterable[Delta]) -> dict[str, Any]:
         delta_contents = []
         delta_client_ids = []
 
@@ -533,7 +533,7 @@ class ProcessProjectfileJobRun(JobRun):
         "%(project__project_filename)s",
     ]
 
-    def get_context(self, *args) -> Dict[str, Any]:
+    def get_context(self, *args) -> dict[str, Any]:
         context = super().get_context(*args)
 
         if not context.get("project__project_filename"):
@@ -573,7 +573,7 @@ def after_docker_exception(self) -> None:
 def cancel_orphaned_workers():
     client: DockerClient = docker.from_env()
 
-    running_workers: List[Container] = client.containers.list(
+    running_workers: list[Container] = client.containers.list(
         filters={"label": f"app={settings.ENVIRONMENT}_worker"},
     )
 

docker-qgis/entrypoint.py

@@ -4,7 +4,7 @@
 import logging
 import os
 from pathlib import Path
-from typing import Dict, Union
+from typing import Union
 
 import qfieldcloud.qgis.apply_deltas
 import qfieldcloud.qgis.process_projectfile
@@ -135,13 +135,13 @@ def _call_qfieldsync_packager(project_filename: Path, package_dir: Path) -> str:
     return packaged_project_filename
 
 
-def _extract_layer_data(project_filename: Union[str, Path]) -> Dict:
+def _extract_layer_data(project_filename: Union[str, Path]) -> dict:
     logging.info("Extracting QGIS project layer data…")
 
     project_filename = str(project_filename)
     project = QgsProject.instance()
     project.read(project_filename)
-    layers_by_id = get_layers_data(project)
+    layers_by_id: dict = get_layers_data(project)
 
     logging.info(
         f"QGIS project layer data\n{layers_data_to_string(layers_by_id)}",
@@ -150,7 +150,7 @@ def _extract_layer_data(project_filename: Union[str, Path]) -> Dict:
     return layers_by_id
 
 
-def cmd_package_project(args):
+def cmd_package_project(args: argparse.Namespace):
     workflow = Workflow(
         id="package_project",
         name="Package Project",
@@ -228,7 +228,7 @@ def cmd_package_project(args):
     )
 
 
-def cmd_apply_deltas(args):
+def cmd_apply_deltas(args: argparse.Namespace):
     workflow = Workflow(
         id="apply_changes",
         name="Apply Changes",
@@ -286,7 +286,7 @@ def cmd_apply_deltas(args):
     )
 
 
-def cmd_process_projectfile(args):
+def cmd_process_projectfile(args: argparse.Namespace):
     workflow = Workflow(
         id="process_projectfile",
         name="Process Projectfile",
@@ -399,5 +399,5 @@ def cmd_process_projectfile(args):
     )
     parser_process_projectfile.set_defaults(func=cmd_process_projectfile)
 
-    args = parser.parse_args()
+    args: argparse.Namespace = parser.parse_args()
     args.func(args)