diff --git a/lib/galaxy/config/__init__.py b/lib/galaxy/config/__init__.py index bedc317eca1..abcf29c25e4 100644 --- a/lib/galaxy/config/__init__.py +++ b/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) diff --git a/lib/galaxy/datatypes/images.py b/lib/galaxy/datatypes/images.py index d77d3484ab9..06f7ee347cb 100644 --- a/lib/galaxy/datatypes/images.py +++ b/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 diff --git a/lib/galaxy/util/__init__.py b/lib/galaxy/util/__init__.py index b8f479c71b8..1e0559418ce 100644 --- a/lib/galaxy/util/__init__.py +++ b/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. diff --git a/lib/galaxy/workflow/modules.py b/lib/galaxy/workflow/modules.py index f259f3a5c8e..d00780a2e82 100644 --- a/lib/galaxy/workflow/modules.py +++ b/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: diff --git a/setup.cfg b/setup.cfg index ac8a0d87bf9..94bde194414 100644 --- a/setup.cfg +++ b/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]