From c39057258fc27507fa29e5df3ff04359084df7cb Mon Sep 17 00:00:00 2001 From: Fabrice Brito Date: Tue, 26 Nov 2024 15:07:36 +0100 Subject: [PATCH] typing List/list for python 3.8 --- zoo_argowf_runner/runner.py | 1 + zoo_argowf_runner/template.py | 33 +++++++++++++++++---------------- zoo_argowf_runner/volume.py | 6 +++--- 3 files changed, 21 insertions(+), 19 deletions(-) diff --git a/zoo_argowf_runner/runner.py b/zoo_argowf_runner/runner.py index a2b2e7a..1296a28 100644 --- a/zoo_argowf_runner/runner.py +++ b/zoo_argowf_runner/runner.py @@ -13,6 +13,7 @@ volume_claim_template, ) + try: import zoo except ImportError: diff --git a/zoo_argowf_runner/template.py b/zoo_argowf_runner/template.py index 1a5c52e..acb1304 100644 --- a/zoo_argowf_runner/template.py +++ b/zoo_argowf_runner/template.py @@ -24,6 +24,7 @@ ) from typing import Optional +from typing import List, Dict def synchronization( @@ -43,8 +44,8 @@ def synchronization( def workflow_step( name: str, - parameters: Optional[list[Parameter]] = None, - artifacts: Optional[list[Artifact]] = None, + parameters: Optional[List[Parameter]] = None, + artifacts: Optional[List[Artifact]] = None, template: Optional[str] = None, template_ref: Optional[TemplateRef] = None, ) -> WorkflowStep: @@ -58,11 +59,11 @@ def workflow_step( def template( name: str, - subStep: Optional[list[WorkflowStep]] = None, - inputs_parameters: Optional[list[dict] | Inputs] = None, - inputs_artifacts: Optional[list[dict] | Inputs] = None, - outputs_parameters: Optional[list[dict] | Outputs] = None, - outputs_artifacts: Optional[list[dict] | Outputs] = None, + subStep: Optional[List[WorkflowStep]] = None, + inputs_parameters: Optional[List[dict] | Inputs] = None, + inputs_artifacts: Optional[List[dict] | Inputs] = None, + outputs_parameters: Optional[List[dict] | Outputs] = None, + outputs_artifacts: Optional[List[dict] | Outputs] = None, script: Optional[ScriptTemplate] = None, ) -> Steps: @@ -75,13 +76,13 @@ def template( inputs = Inputs() outputs = Outputs() - if isinstance(inputs_parameters, list): + if isinstance(inputs_parameters, List): parameters = [Parameter(name=elem["name"]) for elem in inputs_parameters] inputs.parameters = parameters elif isinstance(inputs_parameters, Inputs): inputs = inputs_parameters - if isinstance(inputs_artifacts, list): + if isinstance(inputs_artifacts, List): artifacts = [ Artifact(name=elem["name"], from_expression=elem["from_expression"]) for elem in inputs_artifacts @@ -90,7 +91,7 @@ def template( elif isinstance(inputs_artifacts, Inputs): inputs = inputs_artifacts - if isinstance(outputs_parameters, list): + if isinstance(outputs_parameters, List): if "expression" in outputs_parameters[0].keys(): parameters = [ Parameter( @@ -108,7 +109,7 @@ def template( elif isinstance(outputs_parameters, Outputs): outputs = outputs_parameters - if isinstance(outputs_artifacts, list): + if isinstance(outputs_artifacts, List): artifacts = [ Artifact(name=elem["name"], from_expression=elem["from_expression"]) for elem in outputs_artifacts @@ -136,12 +137,12 @@ def generate_workflow( entrypoint: str, service_account_name: Optional[str] = None, annotations: Optional[dict] = None, - inputs: Optional[list[dict]] = None, + inputs: Optional[List[dict]] = None, synchronization: Optional[Synchronization] = None, - volume_claim_template: Optional[list[PersistentVolumeClaim]] = None, - secret_volume: Optional[list[Volume]] = None, - config_map_volume: Optional[list[Volume]] = None, - templates: Optional[list[Steps]] = None, + volume_claim_template: Optional[List[PersistentVolumeClaim]] = None, + secret_volume: Optional[List[Volume]] = None, + config_map_volume: Optional[List[Volume]] = None, + templates: Optional[List[Steps]] = None, namespace: Optional[str] = None, ): diff --git a/zoo_argowf_runner/volume.py b/zoo_argowf_runner/volume.py index 4930dfe..81b36f5 100644 --- a/zoo_argowf_runner/volume.py +++ b/zoo_argowf_runner/volume.py @@ -1,5 +1,5 @@ # Description: This file contains the functions to create the volume related templates for the Argo workflows. -from typing import Optional +from typing import Optional, List, Dict from hera.workflows.models import ( ConfigMapVolumeSource, @@ -19,7 +19,7 @@ def volume_claim_template( name: str, storageClassName: Optional[str] = None, storageSize: Optional[str] = None, - accessMode: Optional[list[str]] = None, + accessMode: Optional[List[str]] = None, ) -> PersistentVolumeClaim: return PersistentVolumeClaim( metadata=ObjectMeta(name=name), @@ -40,7 +40,7 @@ def secret_volume(name: str, secretName: str) -> Volume: def config_map_volume( - name: str, configMapName: str, items: list[dict], defaultMode: int, optional: bool + name: str, configMapName: str, items: List[dict], defaultMode: int, optional: bool ) -> Volume: keyToPath_items = [] for item in items: