feat: update type annotations from PR comments

Author: m0nhawk

src/argowrapper/engine/argo_engine.py

@@ -1,6 +1,6 @@
 import traceback
 from datetime import datetime
-from typing import Any, Dict, List, Tuple, Union
+from typing import Any, Dict, List, Tuple, Union, Optional
 
 import argo_workflows
 from argo_workflows.api import (
@@ -75,7 +75,7 @@ def __init__(self, dry_run: bool = False):
         )
         self.artifact_api_instance = artifact_service_api.ArtifactServiceApi(api_client)
 
-    def _get_workflow_details_dict(self, workflow_name: Union[str, None]) -> Dict:
+    def _get_workflow_details_dict(self, workflow_name: Optional[str]) -> Dict:
         return self.api_instance.get_workflow(
             namespace=ARGO_NAMESPACE,
             name=workflow_name,
@@ -84,7 +84,7 @@ def _get_workflow_details_dict(self, workflow_name: Union[str, None]) -> Dict:
             _check_return_type=False,
         ).to_dict()
 
-    def _get_archived_workflow_details_dict(self, uid: Union[str, None]) -> Dict:
+    def _get_archived_workflow_details_dict(self, uid: Optional[str]) -> Dict:
         """
         Queries the archived workflows api.
         Raises a argo_workflows.exceptions.NotFoundException if the workflow uid cannot be found
@@ -206,7 +206,7 @@ def _get_lock_for_user(self, username: str) -> Lock:
         return self.user_locks[username]
 
     def get_workflow_details(
-        self, workflow_name: Union[str, None], uid: Union[str, None] = None
+        self, workflow_name: Optional[str], uid: Optional[str] = None
     ) -> Union[Dict[str, Any], str]:
         """
         Gets the workflow status
@@ -381,7 +381,7 @@ def _get_archived_workflow_wf_name_and_team_project(
         return given_name, team_project, gen3username
 
     def get_workflows_for_team_projects_and_user(
-        self, team_projects: List[str], auth_header: Union[str, None]
+        self, team_projects: List[str], auth_header: Optional[str]
     ) -> List[Dict]:
         team_project_workflows = self.get_workflows_for_team_projects(team_projects)
         user_workflows = self.get_workflows_for_user(auth_header)
@@ -418,7 +418,7 @@ def get_workflows_for_team_project(self, team_project: str) -> List[Dict]:
         workflows = self.get_workflows_for_label_selector(label_selector=label_selector)
         return workflows
 
-    def get_workflows_for_user(self, auth_header: Union[str, None]) -> List[Dict]:
+    def get_workflows_for_user(self, auth_header: Optional[str]) -> List[Dict]:
         """
         Get the list of all workflows for the current user. Each item in the list
         contains the workflow name, its status, start and end time.
@@ -602,7 +602,7 @@ def get_workflow_logs(self, workflow_name: str, uid: str) -> List[Dict[str, Any]
                 f"could not get status of {workflow_name}, workflow does not exist"
             )
 
-    def workflow_submission(self, request_body: Dict, auth_header: Union[str, None]):
+    def workflow_submission(self, request_body: Dict, auth_header: Optional[str]):
         # Lock function so only one can run at a time per user
         username = argo_engine_helper.get_username_from_token(auth_header)
         user_lock = self._get_lock_for_user(username)
@@ -711,7 +711,7 @@ def check_user_info_for_billing_id_and_workflow_limit(self, request_token):
             return None, None
 
     def check_user_monthly_workflow_cap(
-        self, request_token, billing_id, custom_limit: Union[int, None]
+        self, request_token, billing_id, custom_limit: Optional[int]
     ):
         """
         Query Argo service to see how many workflow runs user already

src/argowrapper/engine/helpers/argo_engine_helper.py

@@ -2,7 +2,7 @@
 import random
 import re
 import string
-from typing import Any, Callable, Dict, List, Optional, Union
+from typing import Any, Callable, Dict, List, Optional
 
 import jwt
 
@@ -216,7 +216,7 @@ def convert_string_to_pod_label(value: str) -> str:
     return label
 
 
-def get_username_from_token(header_and_or_token: Union[str, None]) -> str:
+def get_username_from_token(header_and_or_token: Optional[str]) -> str:
     """
 
     Args:

src/argowrapper/engine/helpers/workflow_factory.py

@@ -1,4 +1,4 @@
-from typing import Any, Dict, Union
+from typing import Any, Dict, Optional
 
 from argowrapper.constants import WORKFLOW
 from argowrapper.workflows.argo_workflows.gwas import GWAS
@@ -9,7 +9,7 @@ class WorkflowFactory:
     def _get_workflow(
         namespace: str,
         request_body: Dict[str, Any],
-        auth_header: Union[str, None],
+        auth_header: Optional[str],
         workflow_type: WORKFLOW,
     ):
         workflows = {WORKFLOW.GWAS: GWAS}

src/argowrapper/workflows/argo_workflows/gwas.py

@@ -1,4 +1,4 @@
-from typing import Dict, List, Union
+from typing import Dict, List, Optional
 
 import argowrapper.engine.helpers.argo_engine_helper as argo_engine_helper
 from argowrapper import logger
@@ -73,7 +73,7 @@ def __init__(
         self,
         namespace: str,
         request_body: Dict,
-        auth_header: Union[str, None],
+        auth_header: Optional[str],
         dry_run=False,
     ):
         self.username = argo_engine_helper.get_username_from_token(auth_header)