From 5885c49b896de633881edc0b0315bac7902cac35 Mon Sep 17 00:00:00 2001 From: Adrien Date: Fri, 25 Aug 2023 10:57:05 +0200 Subject: [PATCH] Type hints in worker_wrapper (#778) --- docker-app/worker_wrapper/wrapper.py | 26 +++++++++++++------------- docker-qgis/entrypoint.py | 14 +++++++------- 2 files changed, 20 insertions(+), 20 deletions(-) diff --git a/docker-app/worker_wrapper/wrapper.py b/docker-app/worker_wrapper/wrapper.py index a9c5eeb44..f70e97538 100644 --- a/docker-app/worker_wrapper/wrapper.py +++ b/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"}, ) diff --git a/docker-qgis/entrypoint.py b/docker-qgis/entrypoint.py index 3189a4bda..06ec2c7d4 100755 --- a/docker-qgis/entrypoint.py +++ b/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)