diff --git a/files/galaxy_jwd.py b/files/galaxy_jwd.py index 99a5831..9214e8a 100644 --- a/files/galaxy_jwd.py +++ b/files/galaxy_jwd.py @@ -12,7 +12,8 @@ import textwrap from argparse import RawDescriptionHelpFormatter from datetime import datetime -from typing import Optional, Tuple +from pathlib import Path +from typing import Optional, Tuple, Union from xml.dom.minidom import parse import psycopg2 @@ -230,7 +231,7 @@ def main(): print(f"{job_id}: {jwd_path}") -def extract_password_from_pgpass(pgpass_file: str) -> str: +def extract_password_from_pgpass(pgpass_file: Union[str, Path]) -> Union[str, None]: """Extract the password from the ~/.pgpass file. The ~/.pgpass file should have the following format: @@ -257,7 +258,7 @@ def extract_password_from_pgpass(pgpass_file: str) -> str: ) -def get_object_store_conf_path(galaxy_config_file: str) -> str: +def get_object_store_conf_path(galaxy_config_file: Union[str, Path]) -> Union[str, None]: """Get the path to the object_store_conf.xml file. Args: @@ -327,7 +328,7 @@ def parse_object_store_yaml(object_store_conf: str) -> dict: return backends -def get_pulsar_staging_dir(galaxy_pulsar_app_conf: str) -> str: +def get_pulsar_staging_dir(galaxy_pulsar_app_conf: Union[str, Path]) -> str: """Get the path to the pulsar staging directory. Args: @@ -354,11 +355,11 @@ def get_pulsar_staging_dir(galaxy_pulsar_app_conf: str) -> str: def decode_path( - job_id: int, + job_id: Union[int, str], metadata: list, backends_dict: dict, job_runner_name: Optional[str] = None, -) -> str: +) -> Union[str, None]: """Decode the path of JWDs and check if the path exists. Args: diff --git a/files/walle.py b/files/walle.py index 5822de5..9f5433f 100644 --- a/files/walle.py +++ b/files/walle.py @@ -16,7 +16,7 @@ import sys import time import zlib -from typing import Dict, List +from typing import Dict, List, Union import galaxy_jwd import requests @@ -476,7 +476,7 @@ def scan_file_for_malware( return matches -def construct_malware_list(malware_yaml: dict) -> [Malware]: +def construct_malware_list(malware_yaml: dict) -> list[Malware]: """ creates a flat list of malware objects, that hold all info The nested structure in yaml is for better optical structuring @@ -537,7 +537,13 @@ def get_jwd_path(self, job: Job) -> str: class RunningJobDatabase(galaxy_jwd.Database): - def __init__(self, db_name: str, db_host=None, db_user=None, db_password=None): + def __init__( + self, + db_name: str, + db_host: Union[str, None] = None, + db_user: Union[str, None] = None, + db_password: Union[str, None] = None, + ): super().__init__( db_name, db_user,