Source code for galaxy.managers.cloudauthzs

Manager and (de)serializer for cloud authorizations (cloudauthzs).

import logging
from typing import Dict

from galaxy import model
from galaxy.exceptions import InternalServerError
from galaxy.managers import (

log = logging.getLogger(__name__)

[docs]class CloudAuthzManager(sharable.SharableModelManager): model_class = model.CloudAuthz foreign_key_name = "cloudauthz"
[docs]class CloudAuthzsSerializer(base.ModelSerializer): """ Interface/service object for serializing cloud authorizations (cloudauthzs) into dictionaries. """ model_manager_class = CloudAuthzManager
[docs] def __init__(self, app, **kwargs): super().__init__(app, **kwargs) self.cloudauthzs_manager = self.manager self.default_view = "summary" self.add_view( "summary", [ "id", "model_class", "user_id", "provider", "config", "authn_id", "last_update", "last_activity", "create_time", "description", ], )
[docs] def add_serializers(self): super().add_serializers() # Arguments of the following lambda functions: # i : an instance of galaxy.model.CloudAuthz. # k : serialized dictionary key (e.g., 'model_class', 'provider'). # **c: a dictionary containing 'trans' and 'user' objects. serializers: Dict[str, base.Serializer] = { "id": lambda item, key, **context:, "model_class": lambda item, key, **context: "CloudAuthz", "user_id": lambda item, key, **context:, "provider": lambda item, key, **context: str(item.provider), "config": lambda item, key, **context: item.config, "authn_id": lambda item, key, **context: 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: item.create_time.isoformat(), "description": lambda item, key, **context: str(item.description), } self.serializers.update(serializers)
[docs]class CloudAuthzsDeserializer(base.ModelDeserializer): """ Service object for validating and deserializing dictionaries that update/alter cloudauthz configurations. """ model_manager_class = CloudAuthzManager
[docs] def add_deserializers(self): super().add_deserializers() self.deserializers.update( { "authn_id": self.deserialize_and_validate_authn_id, "provider": self.default_deserializer, "config": self.default_deserializer, "description": self.default_deserializer, } )
[docs] def deserialize_and_validate_authn_id(self, item, key, val, **context): """ Deserializes an authentication ID (authn_id), and asserts if the current user can assume that authentication. :type item: galaxy.model.CloudAuthz :param item: an instance of cloudauthz :type key: string :param key: `authn_id` attribute of the cloudauthz object (i.e., the `item` param). :type val: string :param val: the value of `authn_id` attribute of the cloudauthz object (i.e., the `item` param). :type context: dict :param context: a dictionary object containing Galaxy `trans`. :rtype: string :return: decoded authentication ID. """ decoded_authn_id =, object_name="authz") trans = context.get("trans") if trans is None: log.debug("Not found expected `trans` when deserializing CloudAuthz.") raise InternalServerError try:, decoded_authn_id) except Exception as e: raise e return decoded_authn_id