diff --git a/lib/galaxy/managers/base.py b/lib/galaxy/managers/base.py index 61ebb5f162d..baf6be8e8e8 100644 --- a/lib/galaxy/managers/base.py +++ b/lib/galaxy/managers/base.py @@ -30,7 +30,6 @@ import logging import re from typing import ( Any, - Callable, Dict, List, NamedTuple, @@ -547,6 +546,12 @@ class SkipAttribute(Exception): """ +class Serializer(Protocol): + + def __call__(self, item: Any, key: str, **context) -> Any: + ... + + class ModelSerializer(HasAModelManager): """ Turns models into JSONable dicts. @@ -581,7 +586,7 @@ class ModelSerializer(HasAModelManager): # that is in neither serializable_keyset or serializers self.serializable_keyset: Set[str] = set() # a map of dictionary keys to the functions (often lambdas) that create the values for those keys - self.serializers: Dict[str, Callable] = {} + self.serializers: Dict[str, Serializer] = {} # add subclass serializers defined there self.add_serializers() # update the keyset by the serializers (removing the responsibility from subclasses) @@ -662,14 +667,14 @@ class ModelSerializer(HasAModelManager): return getattr(item, key) # serializers for common galaxy objects - def serialize_date(self, item, key, **context): + def serialize_date(self, item: Any, key: str, **context): """ Serialize a date attribute of `item`. """ date = getattr(item, key) return date.isoformat() if date is not None else None - def serialize_id(self, item, key, **context): + def serialize_id(self, item: Any, key: str, **context): """ Serialize an id attribute of `item`. """ @@ -677,7 +682,7 @@ class ModelSerializer(HasAModelManager): # Note: it may not be best to encode the id at this layer return self.app.security.encode_id(id) if id is not None else None - def serialize_type_id(self, item, key, **context): + def serialize_type_id(self, item: Any, key: str, **context): """ Serialize an type-id for `item`. """ diff --git a/lib/galaxy/managers/cloudauthzs.py b/lib/galaxy/managers/cloudauthzs.py index e568fe62924..005fbe780eb 100644 --- a/lib/galaxy/managers/cloudauthzs.py +++ b/lib/galaxy/managers/cloudauthzs.py @@ -3,6 +3,7 @@ Manager and (de)serializer for cloud authorizations (cloudauthzs). """ import logging +from typing import Dict from galaxy import model from galaxy.exceptions import ( @@ -51,18 +52,19 @@ class CloudAuthzsSerializer(base.ModelSerializer): # i : an instance of galaxy.model.CloudAuthz. # k : serialized dictionary key (e.g., 'model_class', 'provider'). # **c: a dictionary containing 'trans' and 'user' objects. - self.serializers.update({ - 'id': lambda i, k, **c: self.app.security.encode_id(i.id), - 'model_class': lambda *a, **c: 'CloudAuthz', - 'user_id': lambda i, k, **c: self.app.security.encode_id(i.user_id), - 'provider': lambda i, k, **c: str(i.provider), - 'config': lambda i, k, **c: i.config, - 'authn_id': lambda i, k, **c: self.app.security.encode_id(i.authn_id) if i.authn_id else None, - 'last_update': lambda i, k, **c: str(i.last_update), - 'last_activity': lambda i, k, **c: str(i.last_activity), - 'create_time': lambda i, k, **c: str(i.create_time), - 'description': lambda i, k, **c: str(i.description) - }) + serializers: Dict[str, base.Serializer] = { + 'id': lambda item, key, **context: self.app.security.encode_id(item.id), + 'model_class': lambda item, key, **context: 'CloudAuthz', + 'user_id': lambda item, key, **context: self.app.security.encode_id(item.user_id), + 'provider': lambda item, key, **context: str(item.provider), + 'config': lambda item, key, **context: item.config, + 'authn_id': lambda item, key, **context: self.app.security.encode_id(item.authn_id) if item.authn_id else None, + 'last_update': lambda item, key, **context: str(item.last_update), + 'last_activity': lambda item, key, **context: str(item.last_activity), + 'create_time': lambda item, key, **context: str(item.create_time), + 'description': lambda item, key, **context: str(item.description) + } + self.serializers.update(serializers) class CloudAuthzsDeserializer(base.ModelDeserializer): diff --git a/lib/galaxy/managers/configuration.py b/lib/galaxy/managers/configuration.py index c3f77ff8594..27427d89606 100644 --- a/lib/galaxy/managers/configuration.py +++ b/lib/galaxy/managers/configuration.py @@ -116,18 +116,18 @@ class ConfigSerializer(base.ModelSerializer): def add_serializers(self): - def _defaults_to(default): - return lambda config, key, **context: getattr(config, key, default) + def _defaults_to(default) -> base.Serializer: + return lambda item, key, **context: getattr(item, key, default) - def _use_config(config, key, **context): + def _use_config(item, key: str, **context): """Let config object determine the value for key""" - assert hasattr(config, key) - return config.config_value_for_host(key, context.get("host")) + assert hasattr(item, key) + return item.config_value_for_host(key, context.get("host")) - def _config_is_truthy(config, key, **context): - return True if config.get(key) else False + def _config_is_truthy(item, key, **context): + return True if item.get(key) else False - self.serializers = { + self.serializers: Dict[str, base.Serializer] = { # TODO: this is available from user data, remove 'is_admin_user': lambda *a, **c: False, 'brand': _use_config, @@ -167,7 +167,7 @@ class ConfigSerializer(base.ModelSerializer): 'ga_code': _use_config, 'plausible_server': _use_config, 'plausible_domain': _use_config, - 'markdown_to_pdf_available': lambda config, key, **context: weasyprint_available(), + 'markdown_to_pdf_available': lambda item, key, **context: weasyprint_available(), 'matomo_server': _use_config, 'matomo_site_id': _use_config, 'enable_unique_workflow_defaults': _use_config, @@ -180,7 +180,7 @@ class ConfigSerializer(base.ModelSerializer): # TODO: is there no 'correct' way to get an api url? controller='api', action='tools' is a hack # at any rate: the following works with path_prefix but is still brittle # TODO: change this to (more generic) upload_path and incorporate config.nginx_upload_path into building it - 'nginx_upload_path': lambda config, key, **context: getattr(config, key, False), + 'nginx_upload_path': lambda item, key, **context: getattr(item, key, False), 'chunk_upload_size': _use_config, 'ftp_upload_site': _use_config, 'version_major': _defaults_to(None), @@ -193,22 +193,22 @@ class ConfigSerializer(base.ModelSerializer): 'message_box_content': _use_config, 'message_box_visible': _use_config, 'message_box_class': _use_config, - 'server_startttime': lambda config, key, **context: server_starttime, + 'server_startttime': lambda item, key, **context: server_starttime, 'mailing_join_addr': _defaults_to('galaxy-announce-join@bx.psu.edu'), # should this be the schema default? - 'server_mail_configured': lambda config, key, **context: bool(config.smtp_server), + 'server_mail_configured': lambda item, key, **context: bool(item.smtp_server), 'registration_warning_message': _use_config, 'welcome_url': _use_config, 'show_welcome_with_login': _defaults_to(True), # schema default is False 'cookie_domain': _use_config, 'python': _defaults_to((sys.version_info.major, sys.version_info.minor)), 'select_type_workflow_threshold': _use_config, - 'file_sources_configured': lambda config, key, **context: self.app.file_sources.custom_sources_configured, - 'panel_views': lambda config, key, **content: self.app.toolbox.panel_view_dicts(), + 'file_sources_configured': lambda item, key, **context: self.app.file_sources.custom_sources_configured, + 'panel_views': lambda item, key, **context: self.app.toolbox.panel_view_dicts(), 'default_panel_view': _use_config, 'upload_from_form_button': _use_config, 'release_doc_base_url': _use_config, 'expose_user_email': _use_config, - 'user_library_import_dir_available': lambda config, key, **context: bool(config.get('user_library_import_dir')), + 'user_library_import_dir_available': lambda item, key, **context: bool(item.get('user_library_import_dir')), 'welcome_directory': _use_config, } diff --git a/lib/galaxy/managers/datasets.py b/lib/galaxy/managers/datasets.py index 7dc7d1d4505..528e149efaf 100644 --- a/lib/galaxy/managers/datasets.py +++ b/lib/galaxy/managers/datasets.py @@ -4,7 +4,7 @@ Manager and Serializer for Datasets. import glob import logging import os -from typing import List, Type +from typing import Dict, List, Type from galaxy import ( exceptions, @@ -171,25 +171,26 @@ class DatasetSerializer(base.ModelSerializer, deletable.PurgableSerializerMixin) def add_serializers(self): super().add_serializers() deletable.PurgableSerializerMixin.add_serializers(self) - - self.serializers.update({ + serializers: Dict[str, base.Serializer] = { 'create_time': self.serialize_date, 'update_time': self.serialize_date, - 'uuid': lambda i, k, **c: str(i.uuid) if i.uuid else None, + 'uuid': lambda item, key, **context: str(item.uuid) if item.uuid else None, 'file_name': self.serialize_file_name, 'extra_files_path': self.serialize_extra_files_path, 'permissions': self.serialize_permissions, - 'total_size': lambda i, k, **c: int(i.get_total_size()), - 'file_size': lambda i, k, **c: int(i.get_size()) - }) + 'total_size': lambda item, key, **context: int(item.get_total_size()), + 'file_size': lambda item, key, **context: int(item.get_size()) + } + self.serializers.update(serializers) - def serialize_file_name(self, dataset, key, user=None, **context): + def serialize_file_name(self, item, key, user=None, **context): """ If the config allows or the user is admin, return the file name of the file that contains this dataset's data. """ + dataset = item is_admin = self.user_manager.is_admin(user, trans=context.get("trans")) # expensive: allow config option due to cost of operation if is_admin or self.app.config.expose_dataset_path: @@ -197,10 +198,11 @@ class DatasetSerializer(base.ModelSerializer, deletable.PurgableSerializerMixin) return dataset.file_name self.skip() - def serialize_extra_files_path(self, dataset, key, user=None, **context): + def serialize_extra_files_path(self, item, key, user=None, **context): """ If the config allows or the user is admin, return the file path. """ + dataset = item is_admin = self.user_manager.is_admin(user, trans=context.get("trans")) # expensive: allow config option due to cost of operation if is_admin or self.app.config.expose_dataset_path: @@ -208,9 +210,10 @@ class DatasetSerializer(base.ModelSerializer, deletable.PurgableSerializerMixin) return dataset.extra_files_path self.skip() - def serialize_permissions(self, dataset, key, user=None, **context): + def serialize_permissions(self, item, key, user=None, **context): """ """ + dataset = item trans = context.get("trans") if not self.dataset_manager.permissions.manage.is_permitted(dataset, user, trans=trans): self.skip() @@ -451,30 +454,30 @@ class _UnflattenedMetadataDatasetAssociationSerializer(base.ModelSerializer, super().add_serializers() deletable.PurgableSerializerMixin.add_serializers(self) - self.serializers.update({ + serializers: Dict[str, base.Serializer] = { 'create_time': self.serialize_date, 'update_time': self.serialize_date, # underlying dataset - 'dataset': lambda i, k, **c: self.dataset_serializer.serialize_to_view(i.dataset, view='summary', **c), - 'dataset_id': self._proxy_to_dataset(key='id'), + 'dataset': lambda item, key, **context: self.dataset_serializer.serialize_to_view(item.dataset, view='summary', **context), + 'dataset_id': self._proxy_to_dataset(proxy_key='id'), # TODO: why is this named uuid!? The da doesn't have a uuid - it's the underlying dataset's uuid! - 'uuid': self._proxy_to_dataset(key='uuid'), + 'uuid': self._proxy_to_dataset(proxy_key='uuid'), # 'dataset_uuid': self._proxy_to_dataset( key='uuid' ), 'file_name': self._proxy_to_dataset(serializer=self.dataset_serializer.serialize_file_name), 'extra_files_path': self._proxy_to_dataset(serializer=self.dataset_serializer.serialize_extra_files_path), 'permissions': self._proxy_to_dataset(serializer=self.dataset_serializer.serialize_permissions), # TODO: do the sizes proxy accurately/in the same way? - 'size': lambda i, k, **c: int(i.get_size()), - 'file_size': lambda i, k, **c: self.serializers['size'](i, k, **c), - 'nice_size': lambda i, k, **c: i.get_size(nice_size=True), + 'size': lambda item, key, **context: int(item.get_size()), + 'file_size': lambda item, key, **context: self.serializers['size'](item, key, **context), + 'nice_size': lambda item, key, **context: item.get_size(nice_size=True), # common to lddas and hdas - from mapping.py 'copied_from_history_dataset_association_id': self.serialize_id, 'copied_from_library_dataset_dataset_association_id': self.serialize_id, - 'info': lambda i, k, **c: i.info.strip() if isinstance(i.info, str) else i.info, - 'blurb': lambda i, k, **c: i.blurb, - 'peek': lambda i, k, **c: i.display_peek() if i.peek and i.peek != 'no peek' else None, + 'info': lambda item, key, **context: item.info.strip() if isinstance(item.info, str) else item.info, + 'blurb': lambda item, key, **context: item.blurb, + 'peek': lambda item, key, **context: item.display_peek() if item.peek and item.peek != 'no peek' else None, 'meta_files': self.serialize_meta_files, 'metadata': self.serialize_metadata, @@ -483,38 +486,40 @@ class _UnflattenedMetadataDatasetAssociationSerializer(base.ModelSerializer, 'rerunnable': self.serialize_rerunnable, 'parent_id': self.serialize_id, - 'designation': lambda i, k, **c: i.designation, + 'designation': lambda item, key, **context: item.designation, # 'extended_metadata': self.serialize_extended_metadata, # 'extended_metadata_id': self.serialize_id, # remapped - 'genome_build': lambda i, k, **c: i.dbkey, + 'genome_build': lambda item, key, **context: item.dbkey, # derived (not mapped) attributes - 'data_type': lambda i, k, **c: f"{i.datatype.__class__.__module__}.{i.datatype.__class__.__name__}", + 'data_type': lambda item, key, **context: f"{item.datatype.__class__.__module__}.{item.datatype.__class__.__name__}", 'converted': self.serialize_converted_datasets, # TODO: metadata/extra files - }) + } + self.serializers.update(serializers) # this an abstract superclass, so no views created # because of that: we need to add a few keys that will use the default serializer self.serializable_keyset.update(['name', 'state', 'tool_version', 'extension', 'visible', 'dbkey']) - def _proxy_to_dataset(self, serializer=None, key=None): + def _proxy_to_dataset(self, serializer: base.Serializer = None, proxy_key=None): # dataset associations are (rough) proxies to datasets - access their serializer using this remapping fn # remapping done by either kwarg key: IOW dataset attr key (e.g. uuid) # or by kwarg serializer: a function that's passed in (e.g. permissions) - if key: - serializer = self.dataset_serializer.serializers.get(key) + if proxy_key: + serializer = self.dataset_serializer.serializers.get(proxy_key) if serializer: - return lambda i, k, **c: serializer(i.dataset, key or k, **c) + return lambda item, key, **context: serializer(item.dataset, proxy_key or key, **context) raise TypeError('kwarg serializer or key needed') - def serialize_meta_files(self, dataset_assoc, key, **context): + def serialize_meta_files(self, item, key, **context): """ Cycle through meta files and return them as a list of dictionaries. """ + dataset_assoc = item meta_files = [] for meta_type in dataset_assoc.metadata_file_types: if getattr(dataset_assoc.metadata, meta_type, None): @@ -526,10 +531,11 @@ class _UnflattenedMetadataDatasetAssociationSerializer(base.ModelSerializer, metadata_file=meta_type))) return meta_files - def serialize_metadata(self, dataset_assoc, key, excluded=None, **context): + def serialize_metadata(self, item, key, excluded=None, **context): """ Cycle through metadata and return as dictionary. """ + dataset_assoc = item # dbkey is a repeat actually (metadata_dbkey == genome_build) # excluded = [ 'dbkey' ] if excluded is None else excluded excluded = [] if excluded is None else excluded @@ -555,34 +561,37 @@ class _UnflattenedMetadataDatasetAssociationSerializer(base.ModelSerializer, return metadata - def serialize_creating_job(self, dataset, key, **context): + def serialize_creating_job(self, item, key, **context): """ Return the id of the Job that created this dataset (or its original) or None if no `creating_job` is found. """ + dataset = item if dataset.creating_job: return self.serialize_id(dataset.creating_job, 'id') else: return None - def serialize_rerunnable(self, dataset, key, **context): + def serialize_rerunnable(self, item, key, **context): """ Return False if this tool that created this dataset can't be re-run (e.g. upload). """ + dataset = item if dataset.creating_job: tool = self.app.toolbox.get_tool(dataset.creating_job.tool_id, dataset.creating_job.tool_version) if tool and tool.is_workflow_compatible: return True return False - def serialize_converted_datasets(self, dataset_assoc, key, **context): + def serialize_converted_datasets(self, item, key, **context): """ Return a file extension -> converted dataset encoded id map with all the existing converted datasets associated with this instance. This filters out deleted associations. """ + dataset_assoc = item id_map = {} for converted in dataset_assoc.implicitly_converted_datasets: if not converted.deleted and converted.dataset: diff --git a/lib/galaxy/managers/hdas.py b/lib/galaxy/managers/hdas.py index fd7d2867e5b..4d19b41281a 100644 --- a/lib/galaxy/managers/hdas.py +++ b/lib/galaxy/managers/hdas.py @@ -18,6 +18,7 @@ from galaxy import ( ) from galaxy.managers import ( annotatable, + base, datasets, secured, taggable, @@ -398,10 +399,10 @@ class HDASerializer( # datasets._UnflattenedMetadataDatasetAssociationSerialize taggable.TaggableSerializerMixin.add_serializers(self) annotatable.AnnotatableSerializerMixin.add_serializers(self) - self.serializers.update({ - 'model_class': lambda *a, **c: 'HistoryDatasetAssociation', - 'history_content_type': lambda *a, **c: 'dataset', - 'hda_ldda': lambda *a, **c: 'hda', + serializers: Dict[str, base.Serializer] = { + 'model_class': lambda item, key, **context: 'HistoryDatasetAssociation', + 'history_content_type': lambda item, key, **context: 'dataset', + 'hda_ldda': lambda item, key, **context: 'hda', 'type_id': self.serialize_type_id, 'copied_from_ldda_id': self.serialize_copied_from_ldda_id, 'history_id': self.serialize_id, @@ -411,7 +412,7 @@ class HDASerializer( # datasets._UnflattenedMetadataDatasetAssociationSerialize 'misc_blurb': self._remap_from('blurb'), 'file_ext': self._remap_from('extension'), 'file_path': self._remap_from('file_name'), - 'resubmitted': lambda i, k, **c: self.hda_manager.has_been_resubmitted(i), + 'resubmitted': lambda item, key, **context: self.hda_manager.has_been_resubmitted(item), 'display_apps': self.serialize_display_apps, 'display_types': self.serialize_old_display_applications, 'visualizations': self.serialize_visualization_links, @@ -420,22 +421,23 @@ class HDASerializer( # datasets._UnflattenedMetadataDatasetAssociationSerialize # TODO: this intermittently causes a routes.GenerationException - temp use the legacy route to prevent this # see also: https://trello.com/c/5d6j4X5y # see also: https://sentry.galaxyproject.org/galaxy/galaxy-main/group/20769/events/9352883/ - 'url': lambda i, k, **c: self.url_for('history_content', - history_id=self.app.security.encode_id(i.history_id), - id=self.app.security.encode_id(i.id)), + 'url': lambda item, key, **context: self.url_for('history_content', + history_id=self.app.security.encode_id(item.history_id), + id=self.app.security.encode_id(item.id)), 'urls': self.serialize_urls, # TODO: backwards compat: need to go away - 'download_url': lambda i, k, **c: self.url_for('history_contents_display', - history_id=self.app.security.encode_id(i.history.id), - history_content_id=self.app.security.encode_id(i.id)), + 'download_url': lambda item, key, **context: self.url_for('history_contents_display', + history_id=self.app.security.encode_id(item.history.id), + history_content_id=self.app.security.encode_id(item.id)), 'parent_id': self.serialize_id, # TODO: to DatasetAssociationSerializer - 'accessible': lambda i, k, user=None, **c: self.manager.is_accessible(i, user, **c), - 'api_type': lambda *a, **c: 'file', - 'type': lambda *a, **c: 'file', - 'created_from_basename': lambda i, k, **c: i.created_from_basename, - }) + 'accessible': lambda item, key, user=None, **c: self.manager.is_accessible(item, user, **c), + 'api_type': lambda item, key, **context: 'file', + 'type': lambda item, key, **context: 'file', + 'created_from_basename': lambda item, key, **context: item.created_from_basename, + } + self.serializers.update(serializers) def serialize(self, hda, keys, user=None, **context): """ @@ -446,10 +448,11 @@ class HDASerializer( # datasets._UnflattenedMetadataDatasetAssociationSerialize keys = self._view_to_keys('inaccessible') return super().serialize(hda, keys, user=user, **context) - def serialize_display_apps(self, hda, key, trans=None, **context): + def serialize_display_apps(self, item, key, trans=None, **context): """ Return dictionary containing new-style display app urls. """ + hda = item display_apps: List[Dict[str, Any]] = [] for display_app in hda.get_display_applications(trans).values(): @@ -465,10 +468,11 @@ class HDASerializer( # datasets._UnflattenedMetadataDatasetAssociationSerialize return display_apps - def serialize_old_display_applications(self, hda, key, trans=None, **context): + def serialize_old_display_applications(self, item, key, trans=None, **context): """ Return dictionary containing old-style display app urls. """ + hda = item display_apps: List[Dict[str, Any]] = [] if not self.app.config.enable_old_display_applications: return display_apps @@ -492,20 +496,22 @@ class HDASerializer( # datasets._UnflattenedMetadataDatasetAssociationSerialize return display_apps - def serialize_visualization_links(self, hda, key, trans=None, **context): + def serialize_visualization_links(self, item, key, trans=None, **context): """ Return a list of dictionaries with links to visualization pages for those visualizations that apply to this hda. """ + hda = item # use older system if registry is off in the config if not self.app.visualizations_registry: return hda.get_visualizations() return self.app.visualizations_registry.get_visualizations(trans, hda) - def serialize_urls(self, hda, key, **context): + def serialize_urls(self, item, key, **context): """ Return web controller urls useful for this HDA. """ + hda = item url_for = self.url_for encoded_id = self.app.security.encode_id(hda.id) urls = { diff --git a/lib/galaxy/managers/hdcas.py b/lib/galaxy/managers/hdcas.py index e72aa46add3..0c21479bafc 100644 --- a/lib/galaxy/managers/hdcas.py +++ b/lib/galaxy/managers/hdcas.py @@ -300,34 +300,33 @@ class HDCASerializer( super().add_serializers() taggable.TaggableSerializerMixin.add_serializers(self) annotatable.AnnotatableSerializerMixin.add_serializers(self) - - self.serializers.update({ - 'model_class': lambda *a, **c: self.hdca_manager.model_class.__class__.__name__, + serializers: Dict[str, base.Serializer] = { + 'model_class': lambda item, key, **context: self.hdca_manager.model_class.__class__.__name__, # TODO: remove - 'type': lambda *a, **c: 'collection', + 'type': lambda item, key, **context: 'collection', # part of a history and container 'history_id': self.serialize_id, - 'history_content_type': lambda *a, **c: self.hdca_manager.model_class.content_type, + 'history_content_type': lambda item, key, **context: self.hdca_manager.model_class.content_type, 'type_id': self.serialize_type_id, 'job_source_id': self.serialize_id, - - 'url': lambda i, k, **c: self.url_for('history_content_typed', - history_id=self.app.security.encode_id(i.history_id), - id=self.app.security.encode_id(i.id), - type=self.hdca_manager.model_class.content_type), + 'url': lambda item, key, **context: self.url_for('history_content_typed', + history_id=self.app.security.encode_id(item.history_id), + id=self.app.security.encode_id(item.id), + type=self.hdca_manager.model_class.content_type), 'contents_url': self.generate_contents_url, 'job_state_summary': self.serialize_job_state_summary - }) + } + self.serializers.update(serializers) - def generate_contents_url(self, hdca, key, **context): + def generate_contents_url(self, item, key, **context): encode_id = self.app.security.encode_id contents_url = self.url_for('contents_dataset_collection', - hdca_id=encode_id(hdca.id), - parent_id=encode_id(hdca.collection_id)) + hdca_id=encode_id(item.id), + parent_id=encode_id(item.collection_id)) return contents_url - def serialize_job_state_summary(self, hdca, key, **context): - states = hdca.job_state_summary.__dict__.copy() + def serialize_job_state_summary(self, item, key, **context): + states = item.job_state_summary.__dict__.copy() del states['_sa_instance_state'] del states['hdca_id'] return states diff --git a/lib/galaxy/managers/history_contents.py b/lib/galaxy/managers/history_contents.py index 323d6f6ae68..3c0dcbc2072 100644 --- a/lib/galaxy/managers/history_contents.py +++ b/lib/galaxy/managers/history_contents.py @@ -450,19 +450,19 @@ class HistoryContentsSerializer(base.ModelSerializer, deletable.PurgableSerializ def add_serializers(self): super().add_serializers() deletable.PurgableSerializerMixin.add_serializers(self) - - self.serializers.update({ + serializers: Dict[str, Serializer] = { 'type_id': self.serialize_type_id, 'history_id': self.serialize_id, 'dataset_id': self.serialize_id_or_skip, 'collection_id': self.serialize_id_or_skip, - }) + } + self.serializers.update(serializers) - def serialize_id_or_skip(self, content, key, **context): + def serialize_id_or_skip(self, item: Any, key: str, **context): """Serialize id or skip if attribute with `key` is not present.""" - if not hasattr(content, key): + if not hasattr(item, key): raise base.SkipAttribute('no such attribute') - return self.serialize_id(content, key, **context) + return self.serialize_id(item, key, **context) class HistoryContentsFilters(base.ModelFilterParser, diff --git a/setup.cfg b/setup.cfg index a78f7bb2111..e5339897287 100644 --- a/setup.cfg +++ b/setup.cfg @@ -605,8 +605,6 @@ check_untyped_defs = False check_untyped_defs = False [mypy-galaxy.tools.evaluation] check_untyped_defs = False -[mypy-galaxy.managers.history_contents] -check_untyped_defs = False [mypy-galaxy.jobs.runners] check_untyped_defs = False [mypy-galaxy.jobs]