Use list and dict container type

Author: mvdbeek

src/ephemeris/_config_models.py

@@ -1,7 +1,5 @@
 from pathlib import Path
 from typing import (
-    Dict,
-    List,
     Optional,
     Union,
 )
@@ -22,7 +20,7 @@ class RepositoryInstallTarget(BaseModel):
     tool_shed_url: Optional[str] = None
     tool_panel_section_id: Optional[str] = None
     tool_panel_section_label: Optional[str] = None
-    revisions: Optional[List[str]] = None
+    revisions: Optional[list[str]] = None
     install_tool_dependencies: Optional[bool] = None
     install_repository_dependencies: Optional[bool] = None
     install_resolver_dependencies: Optional[bool] = None
@@ -33,16 +31,16 @@ class RepositoryInstallTargets(BaseModel):
 
     api_key: Optional[str] = None
     galaxy_instance: Optional[str] = None
-    tools: List[RepositoryInstallTarget]
+    tools: list[RepositoryInstallTarget]
 
 
 class DataManager(BaseModel, extra=Extra.forbid):
-    tags: List[str]
+    tags: list[str]
     tool_id: str
 
 
 class DataManagers(RootModel):
-    root: Dict[str, DataManager]
+    root: dict[str, DataManager]
 
 
 class Genome(BaseModel):
@@ -62,15 +60,15 @@ class Genome(BaseModel):
 
     # Description of actions (data managers) to run on target genome.
     indexers: Optional[
-        List[str]
+        list[str]
     ]  # indexers to run - keyed on repository name - see data_managers.yml for how to resolve these to tools
-    skiplist: Optional[List[str]] = (
+    skiplist: Optional[list[str]] = (
         None  # unimplemented: but if we implement classes of indexers, these will be ones to skip
     )
 
 
 class Genomes(BaseModel):
-    genomes: List[Genome]
+    genomes: list[Genome]
 
 
 def _read_yaml(path: StrOrPath):

src/ephemeris/_idc_data_managers_to_tools.py

@@ -7,11 +7,7 @@
 """
 import argparse
 import logging
-from typing import (
-    Dict,
-    List,
-    NamedTuple,
-)
+from typing import NamedTuple
 
 import yaml
 
@@ -32,12 +28,12 @@
 class DataManager(NamedTuple):
     tool_id: str
     repository_name: str
-    tags: List[str]
+    tags: list[str]
 
 
-def read_data_managers_configuration(path: str) -> Dict[str, DataManager]:
+def read_data_managers_configuration(path: str) -> dict[str, DataManager]:
     raw_data_managers = read_data_managers(path)
-    data_managers: Dict[str, DataManager] = {}
+    data_managers: dict[str, DataManager] = {}
     for repository_name, data_manager_configuration in raw_data_managers.root.items():
         data_manager = DataManager(
             tool_id=data_manager_configuration.tool_id,

src/ephemeris/_idc_split_data_manager_genomes.py

@@ -14,8 +14,6 @@
 from typing import (
     Any,
     Callable,
-    Dict,
-    List,
     Optional,
 )
 
@@ -71,7 +69,7 @@ class SplitOptions:
     filters: Filters = Filters()
 
 
-def tool_id_for(indexer: str, data_managers: Dict[str, DataManager], mode: str) -> str:
+def tool_id_for(indexer: str, data_managers: dict[str, DataManager], mode: str) -> str:
     data_manager = data_managers[indexer]
     assert data_manager, f"Could not find a target data manager for indexer name {indexer}"
     tool_shed_guid = data_manager.tool_id
@@ -88,30 +86,30 @@ def tool_id_for(indexer: str, data_managers: Dict[str, DataManager], mode: str)
 
 class RunDataManager(BaseModel):
     id: str
-    items: Optional[List[Any]] = None
-    params: Optional[List[Any]] = None
-    data_table_reload: Optional[List[str]] = None
+    items: Optional[list[Any]] = None
+    params: Optional[list[Any]] = None
+    data_table_reload: Optional[list[str]] = None
 
 
 class RunDataManagers(BaseModel):
-    data_managers: List[RunDataManager]
+    data_managers: list[RunDataManager]
 
 
 class DataManager(BaseModel, extra=Extra.forbid):
-    tags: List[str]
+    tags: list[str]
     tool_id: str
 
 
 class DataManagers(RootModel):
-    root: Dict[str, DataManager]
+    root: dict[str, DataManager]
 
 
 class Genome(BaseModel):
     pass
 
 
 class Genomes(BaseModel):
-    genomes: List[Genome]
+    genomes: list[Genome]
 
 
 def ucsc_description_for_build(requested_build: str) -> str:
@@ -258,7 +256,7 @@ def write_task_file(build_id: str, indexer: str, run_data_manager: RunDataManage
 
 
 class GalaxyHistoryIsBuildComplete:
-    def __init__(self, history_names: List[str]):
+    def __init__(self, history_names: list[str]):
         self._history_names = history_names
 
     def __call__(self, build_id: str, indexer_name: str) -> bool:
@@ -267,7 +265,7 @@ def __call__(self, build_id: str, indexer_name: str) -> bool:
 
 
 class CVMFSPublishIsComplete:
-    def __init__(self, records: Dict[str, List[str]]):
+    def __init__(self, records: dict[str, list[str]]):
         self.records = records
 
     def __call__(self, build_id: str, indexer_name: str) -> bool:
@@ -294,16 +292,16 @@ def _parser():
     return parser
 
 
-def get_galaxy_history_names(args) -> List[str]:
+def get_galaxy_history_names(args) -> list[str]:
     gi = get_galaxy_connection(args, login_required=True)
     return [h["name"] for h in gi.histories.get_histories()]
 
 
-def get_regular_files(dirname: str) -> List[str]:
+def get_regular_files(dirname: str) -> list[str]:
     return [f for f in os.listdir(dirname) if not f.startswith(".")]
 
 
-def get_cvmfs_publish_records(args) -> Dict[str, List[str]]:
+def get_cvmfs_publish_records(args) -> dict[str, list[str]]:
     records = {}
     records_dir = os.path.join(args.cvmfs_root, "record")
     for build_id in get_regular_files(records_dir):

src/ephemeris/generate_tool_list_from_ga_workflow_files.py

@@ -2,10 +2,7 @@
 """Tool to generate tools from workflows"""
 import json
 from argparse import ArgumentParser
-from typing import (
-    Iterable,
-    List,
-)
+from collections.abc import Iterable
 
 import yaml
 
@@ -60,9 +57,9 @@ def get_workflow_dictionary(json_file):
     return mydict
 
 
-def translate_workflow_dictionary_to_tool_list(workflow_dictionary, panel_label: str) -> List[InstallRepoDict]:
+def translate_workflow_dictionary_to_tool_list(workflow_dictionary, panel_label: str) -> list[InstallRepoDict]:
     starting_tool_list = extract_tool_shed_repositories_from_workflow_dict(workflow_dictionary)
-    tool_list: List[InstallRepoDict] = []
+    tool_list: list[InstallRepoDict] = []
     for tool in starting_tool_list:
         sub_dic: InstallRepoDict = {
             "name": tool["name"],
@@ -104,7 +101,7 @@ def print_yaml_tool_list(tool_dictionary, output_file):
     return
 
 
-def reduce_tool_list(tool_list: List[InstallRepoDict]) -> List[InstallRepoDict]:
+def reduce_tool_list(tool_list: list[InstallRepoDict]) -> list[InstallRepoDict]:
     for current_tool in tool_list:
         for tool in tool_list:
             if current_tool is tool:
@@ -121,8 +118,8 @@ def reduce_tool_list(tool_list: List[InstallRepoDict]) -> List[InstallRepoDict]:
     return tool_list
 
 
-def generate_repo_list_from_workflow(workflow_files: Iterable[str], panel_label: str) -> List[InstallRepoDict]:
-    intermediate_tool_list: List[InstallRepoDict] = []
+def generate_repo_list_from_workflow(workflow_files: Iterable[str], panel_label: str) -> list[InstallRepoDict]:
+    intermediate_tool_list: list[InstallRepoDict] = []
     for workflow in workflow_files:
         workflow_dictionary = get_workflow_dictionary(workflow)
         intermediate_tool_list += translate_workflow_dictionary_to_tool_list(workflow_dictionary, panel_label)

src/ephemeris/set_library_permissions.py

@@ -4,7 +4,6 @@
 import argparse
 import logging as log
 import sys
-from typing import List
 
 from bioblend import galaxy
 from rich.progress import Progress
@@ -17,7 +16,7 @@
 # Print iterations progress
 
 
-def get_datasets(gi, library_id) -> List[str]:
+def get_datasets(gi, library_id) -> list[str]:
     objects = gi.libraries.show_dataset(library_id=library_id, dataset_id="")
     datasets = []
     for index in range(len(objects)):

src/ephemeris/shed_tools.py

@@ -41,15 +41,12 @@
 import re
 import time
 from collections import namedtuple
+from collections.abc import Iterable
 from concurrent.futures import (
     thread,
     ThreadPoolExecutor,
 )
-from typing import (
-    Iterable,
-    List,
-    Optional,
-)
+from typing import Optional
 
 import requests
 import yaml
@@ -106,21 +103,21 @@ class InstallRepoDict(TypedDict):
     tool_panel_section_id: NotRequired[Optional[str]]
     tool_panel_section_label: NotRequired[Optional[str]]
     tool_shed_url: NotRequired[str]
-    revisions: NotRequired[List[str]]
+    revisions: NotRequired[list[str]]
     install_repository_dependencies: NotRequired[bool]
     install_resolver_dependencies: NotRequired[bool]
     install_tool_dependencies: NotRequired[bool]
 
 
 class FilterResults(NamedTuple):
-    already_installed_repos: List[InstallRepoDict]
-    not_installed_repos: List[InstallRepoDict]
+    already_installed_repos: list[InstallRepoDict]
+    not_installed_repos: list[InstallRepoDict]
 
 
 class InstallResults(NamedTuple):
-    installed_repositories: List[InstallRepoDict]
-    skipped_repositories: List[InstallRepoDict]
-    errored_repositories: List[InstallRepoDict]
+    installed_repositories: list[InstallRepoDict]
+    skipped_repositories: list[InstallRepoDict]
+    errored_repositories: list[InstallRepoDict]
 
 
 class InstallRepositoryManager:
@@ -131,7 +128,7 @@ def __init__(self, galaxy_instance):
         self.gi = galaxy_instance
         self.tool_shed_client = ToolShedClient(self.gi)
 
-    def installed_repositories(self) -> List[InstallRepoDict]:
+    def installed_repositories(self) -> list[InstallRepoDict]:
         """Get currently installed tools"""
         return GiToToolYaml(
             gi=self.gi,
@@ -142,8 +139,8 @@ def installed_repositories(self) -> List[InstallRepoDict]:
 
     def filter_installed_repos(self, repos: Iterable[InstallRepoDict], check_revision: bool = True) -> FilterResults:
         """This filters a list of repositories"""
-        not_installed_repos: List[InstallRepoDict] = []
-        already_installed_repos: List[InstallRepoDict] = []
+        not_installed_repos: list[InstallRepoDict] = []
+        already_installed_repos: list[InstallRepoDict] = []
         if check_revision:
             # If we want to check if revisions are equal, flatten the list,
             # so each repository - revision combination has its own entry
@@ -167,7 +164,7 @@ def filter_installed_repos(self, repos: Iterable[InstallRepoDict], check_revisio
 
     def install_repositories(
         self,
-        repositories: List[InstallRepoDict],
+        repositories: list[InstallRepoDict],
         log=log,
         force_latest_revision: bool = False,
         default_toolshed: str = "https://toolshed.g2.bx.psu.edu/",
@@ -177,9 +174,9 @@ def install_repositories(
     ):
         """Install a list of tools on the current galaxy"""
         installation_start = dt.datetime.now()
-        installed_repositories: List[InstallRepoDict] = []
-        skipped_repositories: List[InstallRepoDict] = []
-        errored_repositories: List[InstallRepoDict] = []
+        installed_repositories: list[InstallRepoDict] = []
+        skipped_repositories: list[InstallRepoDict] = []
+        errored_repositories: list[InstallRepoDict] = []
         counter = 0
 
         # Check repos for invalid keys
@@ -194,7 +191,7 @@ def install_repositories(
         total_num_repositories = len(flattened_repos)
 
         # Complete the repo information, and make sure each repository has a revision
-        repository_list: List[InstallRepoDict] = []
+        repository_list: list[InstallRepoDict] = []
         for repository in flattened_repos:
             start = dt.datetime.now()
             try:
@@ -651,7 +648,7 @@ def log_repository_install_start(
     )
 
 
-def args_to_repos(args) -> List[InstallRepoDict]:
+def args_to_repos(args) -> list[InstallRepoDict]:
     if args.tool_list_file:
         tool_list = load_yaml_file(args.tool_list_file)
         repos = tool_list["tools"]

src/ephemeris/shed_tools_methods.py

@@ -1,8 +1,5 @@
-from typing import (
-    Iterable,
-    List,
-    TYPE_CHECKING,
-)
+from collections.abc import Iterable
+from typing import TYPE_CHECKING
 
 from bioblend.toolshed import ToolShedInstance
 
@@ -86,7 +83,7 @@ def get_changeset_revisions(repository: "InstallRepoDict", force_latest_revision
 
 def flatten_repo_info(
     repositories: Iterable["InstallRepoDict"],
-) -> List["InstallRepoDict"]:
+) -> list["InstallRepoDict"]:
     """
     Flatten the dict containing info about what tools to install.
     The tool definition YAML file allows multiple revisions to be listed for
@@ -101,7 +98,7 @@ def flatten_repo_info(
              values, those will be returned as separate list items.
     """
 
-    flattened_list: List[InstallRepoDict] = []
+    flattened_list: list[InstallRepoDict] = []
     for repo_info in repositories:
         new_repo_info = repo_info.copy()
         if "revisions" in new_repo_info: