Code cleanups from pyupgrade

Makefile

@@ -30,10 +30,10 @@ SPACE := $() $()
 NEVER_PYUPGRADE_PATHS := .venv/ .tox/ lib/galaxy/schema/bco/ \
 	lib/galaxy/schema/drs/ lib/tool_shed_client/schema/trs \
 	scripts/check_python.py tools/ test/functional/tools/cwl_tools/
-PY37_PYUPGRADE_PATHS := lib/galaxy/exceptions/ lib/galaxy/job_metrics/ \
-	lib/galaxy/objectstore/ lib/galaxy/tool_util/ lib/galaxy/util/ \
-	test/unit/job_metrics/ test/unit/objectstore/ test/unit/tool_util/ \
-	test/unit/util/
+PY38_PYUPGRADE_PATHS := lib/galaxy/exceptions/ lib/galaxy/job_metrics/ \
+	lib/galaxy/objectstore/ lib/galaxy/tool_util/ lib/galaxy/tool_util_models/ \
+	lib/galaxy/util/ test/unit/job_metrics/ test/unit/objectstore/ \
+	test/unit/tool_util/ test/unit/tool_util_models/ test/unit/util/
 
 all: help
 	@echo "This makefile is used for building Galaxy's JS client, documentation, and drive the release process. A sensible all target is not implemented."
@@ -62,10 +62,10 @@ format:  ## Format Python code base
 remove-unused-imports:  ## Remove unused imports in Python code base
 	$(IN_VENV) autoflake --in-place --remove-all-unused-imports --recursive --verbose lib/ test/
 
-pyupgrade:  ## Convert older code patterns to Python 3.7/3.9 idiomatic ones
-	ack --type=python -f | grep -v '^$(subst $(SPACE),\|^,$(NEVER_PYUPGRADE_PATHS) $(PY37_PYUPGRADE_PATHS))' | xargs pyupgrade --py39-plus
-	ack --type=python -f | grep -v '^$(subst $(SPACE),\|^,$(NEVER_PYUPGRADE_PATHS) $(PY37_PYUPGRADE_PATHS))' | xargs auto-walrus
-	ack --type=python -f $(PY37_PYUPGRADE_PATHS) | xargs pyupgrade --py37-plus
+pyupgrade:  ## Convert older code patterns to Python 3.8/3.9 idiomatic ones
+	ack --type=python -f | grep -v '^$(subst $(SPACE),\|^,$(NEVER_PYUPGRADE_PATHS) $(PY38_PYUPGRADE_PATHS))' | xargs pyupgrade --py39-plus
+	ack --type=python -f | grep -v '^$(subst $(SPACE),\|^,$(NEVER_PYUPGRADE_PATHS) $(PY38_PYUPGRADE_PATHS))' | xargs auto-walrus
+	ack --type=python -f $(PY38_PYUPGRADE_PATHS) | xargs pyupgrade --py38-plus
 
 docs-slides-ready:
 	test -f plantuml.jar ||  wget http://jaist.dl.sourceforge.net/project/plantuml/plantuml.jar

client/src/api/schema/schema.ts

@@ -18044,8 +18044,8 @@ export interface components {
              */
             url: string;
         };
-        /** RootModel[Dict[str, int]] */
-        RootModel_Dict_str__int__: {
+        /** RootModel[dict[str, int]] */
+        RootModel_dict_str__int__: {
             [key: string]: number;
         };
         /** RulesParameterModel */
@@ -39456,7 +39456,7 @@ export interface operations {
                     [name: string]: unknown;
                 };
                 content: {
-                    "application/json": components["schemas"]["RootModel_Dict_str__int__"];
+                    "application/json": components["schemas"]["RootModel_dict_str__int__"];
                 };
             };
             /** @description Request Error */

lib/galaxy/app.py

@@ -9,10 +9,7 @@
 from typing import (
     Any,
     Callable,
-    Dict,
-    List,
     Optional,
-    Tuple,
 )
 
 from beaker.cache import CacheManager
@@ -188,7 +185,7 @@
 
 
 class HaltableContainer(Container):
-    haltables: List[Tuple[str, Callable]]
+    haltables: list[tuple[str, Callable]]
 
     def __init__(self) -> None:
         super().__init__()
@@ -593,7 +590,7 @@ def __init__(self, configure_logging=True, use_converters=True, use_display_appl
         self.job_config = self._register_singleton(jobs.JobConfiguration)
 
         # Setup infrastructure for short term storage manager.
-        short_term_storage_config_kwds: Dict[str, Any] = {}
+        short_term_storage_config_kwds: dict[str, Any] = {}
         short_term_storage_config_kwds["short_term_storage_directory"] = self.config.short_term_storage_dir
         short_term_storage_default_duration = self.config.short_term_storage_default_duration
         short_term_storage_maximum_duration = self.config.short_term_storage_maximum_duration

lib/galaxy/authnz/custos_authnz.py

@@ -10,7 +10,6 @@
     timedelta,
 )
 from typing import (
-    List,
     Optional,
 )
 from urllib.parse import quote
@@ -59,9 +58,9 @@ class CustosAuthnzConfiguration:
     redirect_uri: str
     ca_bundle: Optional[str]
     pkce_support: bool
-    accepted_audiences: List[str]
+    accepted_audiences: list[str]
     extra_params: Optional[dict]
-    extra_scopes: List[str]
+    extra_scopes: list[str]
     authorization_endpoint: Optional[str]
     token_endpoint: Optional[str]
     end_session_endpoint: Optional[str]
@@ -581,7 +580,7 @@ class _CustosAuthBasedProviderCacheItem:
         oidc_backend_config: dict
         idphint: str
 
-    _CustosAuthBasedProvidersCache: List[_CustosAuthBasedProviderCacheItem] = []
+    _CustosAuthBasedProvidersCache: list[_CustosAuthBasedProviderCacheItem] = []
 
     @staticmethod
     def GetCustosBasedAuthProvider(provider, oidc_config, oidc_backend_config, idphint=None):

lib/galaxy/celery/__init__.py

@@ -9,7 +9,6 @@
 from typing import (
     Any,
     Callable,
-    Dict,
 )
 
 import pebble
@@ -201,7 +200,7 @@ def wrapper(*args, **kwds):
 
 
 def init_celery_app():
-    celery_app_kwd: Dict[str, Any] = {
+    celery_app_kwd: dict[str, Any] = {
         "include": TASKS_MODULES,
         "task_default_queue": DEFAULT_TASK_QUEUE,
         "task_create_missing_queues": True,
@@ -235,7 +234,7 @@ def schedule_task(task, interval):
                 "schedule": interval,
             }
 
-    beat_schedule: Dict[str, Dict[str, Any]] = {}
+    beat_schedule: dict[str, dict[str, Any]] = {}
     schedule_task("prune_history_audit_table", config.history_audit_table_prune_interval)
     schedule_task("cleanup_short_term_storage", config.short_term_storage_cleanup_interval)
 

lib/galaxy/celery/tasks.py

@@ -503,8 +503,7 @@ def send_notification_to_recipients_async(
 @galaxy_task(action="dispatch pending notifications")
 def dispatch_pending_notifications(notification_manager: NotificationManager):
     """Dispatch pending notifications."""
-    count = notification_manager.dispatch_pending_notifications_via_channels()
-    if count:
+    if count := notification_manager.dispatch_pending_notifications_via_channels():
         log.info(f"Successfully dispatched {count} notifications.")
 
 

lib/galaxy/config/__init__.py

@@ -21,10 +21,7 @@
     Any,
     Callable,
     cast,
-    Dict,
-    List,
     Optional,
-    Set,
     SupportsInt,
     TYPE_CHECKING,
     TypeVar,
@@ -74,7 +71,7 @@
 GALAXY_CONFIG_SCHEMA_PATH = GALAXY_SCHEMAS_PATH / "config_schema.yml"
 REPORTS_CONFIG_SCHEMA_PATH = GALAXY_SCHEMAS_PATH / "reports_config_schema.yml"
 TOOL_SHED_CONFIG_SCHEMA_PATH = GALAXY_SCHEMAS_PATH / "tool_shed_config_schema.yml"
-LOGGING_CONFIG_DEFAULT: Dict[str, Any] = {
+LOGGING_CONFIG_DEFAULT: dict[str, Any] = {
     "disable_existing_loggers": False,
     "version": 1,
     "root": {
@@ -151,7 +148,7 @@
 }
 """Default value for logging configuration, passed to :func:`logging.config.dictConfig`"""
 
-DEPENDENT_CONFIG_DEFAULTS: Dict[str, str] = {
+DEPENDENT_CONFIG_DEFAULTS: dict[str, str] = {
     "mulled_resolution_cache_url": "database_connection",
     "citation_cache_url": "database_connection",
     "biotools_service_cache_url": "database_connection",
@@ -241,13 +238,13 @@ def expand_pretty_datetime_format(value):
 class BaseAppConfiguration(HasDynamicProperties):
     # Override in subclasses (optional): {KEY: config option, VALUE: deprecated directory name}
     # If VALUE == first directory in a user-supplied path that resolves to KEY, it will be stripped from that path
-    renamed_options: Optional[Dict[str, str]] = None
-    deprecated_dirs: Dict[str, str] = {}
-    paths_to_check_against_root: Set[str] = (
+    renamed_options: Optional[dict[str, str]] = None
+    deprecated_dirs: dict[str, str] = {}
+    paths_to_check_against_root: set[str] = (
         set()
     )  # backward compatibility: if resolved path doesn't exist, try resolving w.r.t root
-    add_sample_file_to_defaults: Set[str] = set()  # for these options, add sample config files to their defaults
-    listify_options: Set[str] = set()  # values for these options are processed as lists of values
+    add_sample_file_to_defaults: set[str] = set()  # for these options, add sample config files to their defaults
+    listify_options: set[str] = set()  # values for these options are processed as lists of values
     object_store_store_by: str
     shed_tools_dir: str
 
@@ -441,7 +438,7 @@ def _set_alt_paths(self, option, *alt_paths):
                     return path
 
     def _update_raw_config_from_kwargs(self, kwargs):
-        type_converters: Dict[str, Callable[[Any], Union[bool, int, float, str]]] = {
+        type_converters: dict[str, Callable[[Any], Union[bool, int, float, str]]] = {
             "bool": string_as_bool,
             "int": int,
             "float": float,
@@ -603,7 +600,7 @@ class CommonConfigurationMixin:
     """Shared configuration settings code for Galaxy and ToolShed."""
 
     sentry_dsn: str
-    config_dict: Dict[str, str]
+    config_dict: dict[str, str]
 
     @property
     def admin_users(self):
@@ -712,7 +709,7 @@ class GalaxyAppConfiguration(BaseAppConfiguration, CommonConfigurationMixin):
         "tool_config_file",
     }
 
-    allowed_origin_hostnames: List[str]
+    allowed_origin_hostnames: list[str]
     builds_file_path: str
     container_resolvers_config_file: str
     database_connection: str
@@ -729,32 +726,32 @@ class GalaxyAppConfiguration(BaseAppConfiguration, CommonConfigurationMixin):
     len_file_path: str
     manage_dependency_relationships: bool
     monitor_thread_join_timeout: int
-    mulled_channels: List[str]
+    mulled_channels: list[str]
     new_file_path: str
     nginx_upload_store: str
     password_expiration_period: timedelta
     preserve_python_environment: str
     pretty_datetime_format: str
     sanitize_allowlist_file: str
     shed_tool_data_path: str
-    themes: Dict[str, Dict[str, str]]
-    themes_by_host: Dict[str, Dict[str, Dict[str, str]]]
+    themes: dict[str, dict[str, str]]
+    themes_by_host: dict[str, dict[str, dict[str, str]]]
     tool_data_path: str
     tool_dependency_dir: Optional[str]
-    tool_filters: List[str]
-    tool_label_filters: List[str]
+    tool_filters: list[str]
+    tool_label_filters: list[str]
     tool_path: str
-    tool_section_filters: List[str]
-    toolbox_filter_base_modules: List[str]
+    tool_section_filters: list[str]
+    toolbox_filter_base_modules: list[str]
     track_jobs_in_database: bool
     trust_jupyter_notebook_conversion: bool
     tus_upload_store: str
     use_remote_user: bool
     user_library_import_dir_auto_creation: bool
-    user_library_import_symlink_allowlist: List[str]
-    user_tool_filters: List[str]
-    user_tool_label_filters: List[str]
-    user_tool_section_filters: List[str]
+    user_library_import_symlink_allowlist: list[str]
+    user_tool_filters: list[str]
+    user_tool_label_filters: list[str]
+    user_tool_section_filters: list[str]
     visualization_plugins_directory: str
     workflow_resource_params_mapper: str
 
@@ -788,7 +785,7 @@ def config_value_for_host(self, config_option, host):
         val = getattr(self, config_option)
         if config_option in self.schema.per_host_options:
             per_host_option = f"{config_option}_by_host"
-            per_host: Dict[str, Any] = {}
+            per_host: dict[str, Any] = {}
             if per_host_option in self.config_dict:
                 per_host = self.config_dict[per_host_option] or {}
             else:
@@ -800,7 +797,7 @@ def config_value_for_host(self, config_option, host):
 
         return val
 
-    def _process_config(self, kwargs: Dict[str, Any]) -> None:
+    def _process_config(self, kwargs: dict[str, Any]) -> None:
         self._check_database_connection_strings()
         # Backwards compatibility for names used in too many places to fix
         self.datatypes_config = self.datatypes_config_file
@@ -856,7 +853,7 @@ def _process_config(self, kwargs: Dict[str, Any]) -> None:
             self.tool_data_path = self._in_data_dir(self.schema.defaults["tool_data_path"])
         self.builds_file_path = os.path.join(self.tool_data_path, self.builds_file_path)
         self.len_file_path = os.path.join(self.tool_data_path, self.len_file_path)
-        self.oidc: Dict[str, Dict] = {}
+        self.oidc: dict[str, dict] = {}
         self.fixed_delegated_auth: bool = False
         self.integrated_tool_panel_config = self._in_managed_config_dir(self.integrated_tool_panel_config)
         integrated_tool_panel_tracking_directory = kwargs.get("integrated_tool_panel_tracking_directory")
@@ -1469,7 +1466,7 @@ def get_database_engine_options(kwargs, model_prefix=""):
     Allow options for the SQLAlchemy database engine to be passed by using
     the prefix "database_engine_option".
     """
-    conversions: Dict[str, Callable[[Any], Union[bool, int]]] = {
+    conversions: dict[str, Callable[[Any], Union[bool, int]]] = {
         "convert_unicode": string_as_bool,
         "pool_timeout": int,
         "echo": string_as_bool,