improve type hints for lib/galaxy/config/__init__.py.

lib/galaxy/config/__init__.py

@@ -19,7 +19,19 @@ import tempfile
 import threading
 import time
 from datetime import timedelta
-from typing import Dict, Optional, Set
+from typing import (
+    Any,
+    Callable,
+    cast,
+    Dict,
+    List,
+    Optional,
+    Set,
+    SupportsInt,
+    TYPE_CHECKING,
+    TypeVar,
+    Union,
+)
 
 import yaml
 from beaker.cache import CacheManager
@@ -31,6 +43,7 @@ from galaxy.exceptions import ConfigurationError
 from galaxy.model import mapping
 from galaxy.model.database_utils import database_exists
 from galaxy.model.tool_shed_install.migrate.check import create_or_verify_database as tsi_create_or_verify_database
+from galaxy.structured_app import BasicApp
 from galaxy.util import (
     ExecutionTimer,
     listify,
@@ -49,11 +62,18 @@ from galaxy.web.formatting import expand_pretty_datetime_format
 from galaxy.web_stack import get_stack_facts
 from ..version import VERSION_MAJOR, VERSION_MINOR
 
+if TYPE_CHECKING:
+    from galaxy.jobs import JobConfiguration
+    from galaxy.tool_util.deps.containers import ContainerFinder
+    from galaxy.tools import ToolBox
+    from galaxy.tools.cache import ToolCache
+    from galaxy.tools.search import ToolBoxSearch
+
 log = logging.getLogger(__name__)
 
 GALAXY_APP_NAME = 'galaxy'
 GALAXY_CONFIG_SCHEMA_PATH = 'lib/galaxy/webapps/galaxy/config_schema.yml'
-LOGGING_CONFIG_DEFAULT = {
+LOGGING_CONFIG_DEFAULT: Dict[str, Any] = {
     'disable_existing_loggers': False,
     'version': 1,
     'root': {
@@ -113,6 +133,9 @@ def find_root(kwargs):
     return os.path.abspath(kwargs.get('root_dir', '.'))
 
 
+OptStr = TypeVar("OptStr", None, str)
+
+
 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
@@ -121,6 +144,7 @@ class BaseAppConfiguration(HasDynamicProperties):
     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
+    object_store_store_by: str
 
     def __init__(self, **kwargs):
         self._preprocess_kwargs(kwargs)
@@ -305,6 +329,13 @@ class BaseAppConfiguration(HasDynamicProperties):
 
     def _update_raw_config_from_kwargs(self, kwargs):
 
+        type_converters: Dict[str, Callable[[Any], Union[bool, int, float, str]]] = {
+            "bool": string_as_bool,
+            "int": int,
+            "float": float,
+            "str": str,
+        }
+
         def convert_datatype(key, value):
             datatype = self.schema.app_schema[key].get('type')
             # check for `not None` explicitly (value can be falsy)
@@ -337,8 +368,6 @@ class BaseAppConfiguration(HasDynamicProperties):
                 return ','.join(paths)
             return value
 
-        type_converters = {'bool': string_as_bool, 'int': int, 'float': float, 'str': str}
-
         for key, value in kwargs.items():
             if key in self.schema.app_schema:
                 value = convert_datatype(key, value)
@@ -424,28 +453,33 @@ class BaseAppConfiguration(HasDynamicProperties):
             if path != current_value:
                 setattr(self, key, path)  # update if path has changed
 
-    def _in_root_dir(self, path):
+    def _in_root_dir(self, path: OptStr) -> OptStr:
         return self._in_dir(self.root, path)
 
-    def _in_managed_config_dir(self, path):
+    def _in_managed_config_dir(self, path: OptStr) -> OptStr:
         return self._in_dir(self.managed_config_dir, path)
 
-    def _in_config_dir(self, path):
+    def _in_config_dir(self, path: OptStr) -> OptStr:
         return self._in_dir(self.config_dir, path)
 
-    def _in_sample_dir(self, path):
+    def _in_sample_dir(self, path: OptStr) -> OptStr:
         return self._in_dir(self.sample_config_dir, path)
 
-    def _in_data_dir(self, path):
+    def _in_data_dir(self, path: OptStr) -> OptStr:
         return self._in_dir(self.data_dir, path)
 
-    def _in_dir(self, _dir, path):
-        return os.path.join(_dir, path) if path else None
+    def _in_dir(self, _dir: str, path: OptStr) -> OptStr:
+        if path is not None:
+            return os.path.join(_dir, path)
+        return None
 
 
 class CommonConfigurationMixin:
     """Shared configuration settings code for Galaxy and ToolShed."""
 
+    sentry_dsn: str
+    config_dict: Dict[str, str]
+
     @property
     def admin_users(self):
         return self._admin_users
@@ -547,6 +581,41 @@ class GalaxyAppConfiguration(BaseAppConfiguration, CommonConfigurationMixin):
         'tool_data_table_config_path',
         'tool_config_file',
     }
+    database_connection: str
+    tool_path: str
+    tool_data_path: str
+    builds_file_path: str
+    len_file_path: str
+    integrated_tool_panel_config: str
+    toolbox_filter_base_modules: List[str]
+    tool_filters: List[str]
+    tool_label_filters: List[str]
+    tool_section_filters: List[str]
+    user_tool_filters: List[str]
+    user_tool_section_filters: List[str]
+    user_tool_label_filters: List[str]
+    password_expiration_period: timedelta
+    shed_tool_data_path: str
+    hours_between_check: int
+    galaxy_data_manager_data_path: str
+    use_remote_user: bool
+    cluster_files_directory: str
+    preserve_python_environment: str
+    email_from: str
+    workflow_resource_params_mapper: str
+    sanitize_allowlist_file: str
+    allowed_origin_hostnames: List[str]
+    trust_jupyter_notebook_conversion: bool
+    user_library_import_symlink_allowlist: List[str]
+    user_library_import_dir_auto_creation: bool
+    container_resolvers_config_file: str
+    tool_dependency_dir: Optional[str]
+    involucro_path: str
+    mulled_channels: List[str]
+    nginx_upload_store: str
+    pretty_datetime_format: str
+    visualization_plugins_directory: str
+    galaxy_infrastructure_url: str
 
     def __init__(self, **kwargs):
         super().__init__(**kwargs)
@@ -628,7 +697,9 @@ class GalaxyAppConfiguration(BaseAppConfiguration, CommonConfigurationMixin):
         self.user_tool_section_filters = listify(self.user_tool_section_filters, do_strip=True)
         self.has_user_tool_filters = bool(self.user_tool_filters or self.user_tool_label_filters or self.user_tool_section_filters)
 
-        self.password_expiration_period = timedelta(days=int(self.password_expiration_period))
+        self.password_expiration_period = timedelta(
+            days=int(cast(SupportsInt, self.password_expiration_period))
+        )
 
         if self.shed_tool_data_path:
             self.shed_tool_data_path = self._in_root_dir(self.shed_tool_data_path)
@@ -752,7 +823,7 @@ class GalaxyAppConfiguration(BaseAppConfiguration, CommonConfigurationMixin):
             self.involucro_path = self._in_data_dir(os.path.join(target_dir, "involucro"))
         self.involucro_path = self._in_root_dir(self.involucro_path)
         if self.mulled_channels:
-            self.mulled_channels = [c.strip() for c in self.mulled_channels.split(',')]
+            self.mulled_channels = [c.strip() for c in self.mulled_channels.split(",")]  # type: ignore[attr-defined]
 
         default_job_resubmission_condition = kwargs.get('default_job_resubmission_condition', '')
         if not default_job_resubmission_condition.strip():
@@ -934,8 +1005,11 @@ class GalaxyAppConfiguration(BaseAppConfiguration, CommonConfigurationMixin):
         self._set_alt_paths('file_path', self._in_data_dir('files'))  # this is called BEFORE guessing id/uuid
         ID, UUID = 'id', 'uuid'
         if self.is_set('object_store_store_by'):
-            assert self.object_store_store_by in [ID, UUID], f"Invalid value for object_store_store_by [{self.object_store_store_by}]"
-        elif os.path.basename(self.file_path) == 'objects':
+            if self.object_store_store_by not in [ID, UUID]:
+                raise Exception(
+                    f"Invalid value for object_store_store_by [{self.object_store_store_by}]"
+                )
+        elif os.path.basename(self.file_path) == "objects":
             self.object_store_store_by = UUID
         else:
             self.object_store_store_by = ID
@@ -1050,7 +1124,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 = {
+    conversions: Dict[str, Callable[[Any], Union[bool, int]]] = {
         'convert_unicode': string_as_bool,
         'pool_timeout': int,
         'echo': string_as_bool,
@@ -1131,6 +1205,13 @@ def configure_logging(config):
 class ConfiguresGalaxyMixin:
     """Shared code for configuring Galaxy-like app objects."""
 
+    config: GalaxyAppConfiguration
+    tool_cache: "ToolCache"
+    job_config: "JobConfiguration"
+    toolbox: "ToolBox"
+    toolbox_search: "ToolBoxSearch"
+    container_finder: "ContainerFinder"
+
     def _configure_genome_builds(self, data_table_name="__dbkeys__", load_old_style=True):
         self.genome_builds = GenomeBuilds(self, data_table_name=data_table_name, load_old_style=load_old_style)
 
@@ -1164,10 +1245,14 @@ class ConfiguresGalaxyMixin:
         from galaxy.tool_util.deps.dependencies import AppInfo
         import galaxy.tools.search
 
+        if not isinstance(self, BasicApp):
+            raise Exception("Must inherit from BasicApp")
+
         self.citations_manager = CitationsManager(self)
         self.biotools_metadata_source = get_galaxy_biotools_metadata_source(self.config)
 
         from galaxy.managers.tools import DynamicToolManager
+
         self.dynamic_tools_manager = DynamicToolManager(self)
         self._toolbox_lock = threading.RLock()
         self.toolbox = tools.ToolBox(self.config.tool_configs, self.config.tool_path, self)

lib/galaxy/datatypes/images.py

@@ -305,12 +305,13 @@ class Trk(Binary):
             ('header_size', 'i4'),
         ]
         np_dtype = np.dtype(header_def)
-        header = np.ndarray(
-            shape=(),
-            dtype=np_dtype,
-            buffer=header_raw)
-        if header['header_size'] == 1000 and b'TRACK' in header['magic'] and \
-           header['version'] == 2 and len(header['dim']) == 3:
+        header: np.ndarray = np.ndarray(shape=(), dtype=np_dtype, buffer=header_raw)
+        if (
+            header["header_size"] == 1000
+            and b"TRACK" in header["magic"]
+            and header["version"] == 2
+            and len(header["dim"]) == 3
+        ):
             return True
         return False
 

lib/galaxy/util/__init__.py

@@ -959,7 +959,7 @@ def asbool(obj):
     return bool(obj)
 
 
-def string_as_bool(string):
+def string_as_bool(string: str) -> bool:
     if str(string).lower() in ('true', 'yes', 'on', '1'):
         return True
     else:
@@ -985,7 +985,7 @@ def string_as_bool_or_none(string):
         return False
 
 
-def listify(item, do_strip=False):
+def listify(item, do_strip=False) -> typing.List[typing.Any]:
     """
     Make a single item a single item list.
 

lib/galaxy/workflow/modules.py

@@ -8,6 +8,7 @@ from collections import defaultdict
 from typing import Any, cast, Dict, List, Optional, Union
 
 import packaging.version
+from typing_extensions import TypedDict
 
 from galaxy import (
     exceptions,
@@ -614,6 +615,11 @@ def format_param(trans, formats):
     return format_value
 
 
+class InputModuleState(TypedDict, total=False):
+    optional: bool
+    format: List[str]
+
+
 class InputModule(WorkflowModule):
     default_optional = False
 
@@ -662,14 +668,14 @@ class InputModule(WorkflowModule):
 
     def _parse_state_into_dict(self):
         inputs = self.state.inputs
-        rval = {}
+        rval: InputModuleState = {}
         if "optional" in inputs:
             optional = bool(inputs["optional"])
         else:
             optional = self.default_optional
         rval["optional"] = optional
         if "format" in inputs:
-            formats = listify(inputs["format"])
+            formats: Optional[List[str]] = listify(inputs["format"])
         else:
             formats = None
         if formats:

setup.cfg

@@ -581,8 +581,6 @@ check_untyped_defs = False
 check_untyped_defs = False
 [mypy-galaxy.jobs.runners.state_handler_factory]
 check_untyped_defs = False
-[mypy-galaxy.config]
-check_untyped_defs = False
 [mypy-galaxy.queue_worker]
 check_untyped_defs = False
 [mypy-galaxy.tools]