opensource/galaxy
3
0
Fork 0
mirror of https://github.com/galaxyproject/galaxy.git synced 2026-09-24 16:30:27 +08:00
Code Issues Packages Projects Releases Wiki Activity

Define protocol for serializers - mypy check all the serialiers in manager layer.

Browse Source
This commit is contained in:
John Chilton
2021-10-27 13:41:42 -04:00
parent 789d70845d
commit c3ea408eeb
8 changed files with 128 additions and 109 deletions
Download Patch File Download Diff File Expand all files Collapse all files
lib/galaxy/managers/base.py
+10 -5
View File
@@ -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`.
"""
lib/galaxy/managers/cloudauthzs.py
+14 -12
View File
@@ -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):
lib/galaxy/managers/configuration.py
+15 -15
View File
@@ -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,
}
lib/galaxy/managers/datasets.py
+42 -33
View File
@@ -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:
lib/galaxy/managers/hdas.py
+26 -20
View File
@@ -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 = {
lib/galaxy/managers/hdcas.py
+15 -16
View File
@@ -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
lib/galaxy/managers/history_contents.py
+6 -6
View File
@@ -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,
setup.cfg
-2
View File
@@ -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]