from dataclasses import asdict, dataclass from enum import Enum, unique from typing import Optional, Union from urllib.parse import urljoin import structlog from delphi_api.const import ATLAS_HOST from delphi_api.v3.atlas.models import AtlasImageModel, AtlasImageSchema from delphi_api.v3.constants import DSP_APPLE, DSP_SPOTIFY from delphi_api.v3.data_models.postgres_db import Artist, BaseModel, Playlist LOG = structlog.get_logger(__name__) @dataclass class RouteMap: id_field: str route: str @unique class AtlasRouteGroup(Enum): ARTIST = Artist.__name__ PLAYLIST = Playlist.__name__ #: Links model names to :class:`AtlasRouteGroup` enums MODEL_NAME_ENUM_MAP = { Artist.__name__: AtlasRouteGroup.ARTIST, Playlist.__name__: AtlasRouteGroup.PLAYLIST, } #: Map linking :class:`AtlasRouteGroup` to Atlas routes as :class:`RouteMap` objects ATLAS_ROUTE_MAP = { AtlasRouteGroup.ARTIST: RouteMap(id_field=Artist.particip_no.key, route='artists/by_gras_participant_id'), AtlasRouteGroup.PLAYLIST: { DSP_APPLE: RouteMap(id_field=Playlist.dsp_playlist_id.key, route='playlists/by_apple_music_id'), DSP_SPOTIFY: RouteMap(id_field=Playlist.dsp_playlist_id.key, route='playlists/by_spotify_id'), } } class AtlasImageService: """ This static helper class will create an image uri for relevant models dynamically. To add supported models, update the MAP constants above and potentially :meth:`AtlasImageService._get_route_map()` if the new routes are DSP-depedent. """ @classmethod def get_image_data_by_model(cls, model: Union[BaseModel, dict], model_name: str) -> dict: """ Args: model: An instance of a supported data model (or dictionary of one) model_name: string name of the model passed (since model can be a dictionary) Returns: A dictionary of an image object, or an empty dictionary if none found """ dsp = cls._get_model_value(model, 'dsp') dsp_id = cls._get_model_value(dsp, 'dsp_id') data = cls._get_image_data(model=model, model_name=model_name, dsp_id=dsp_id) return data if data else {} @classmethod def _get_image_data(cls, model: Union[BaseModel, dict], model_name: str, dsp_id: str) -> Optional[dict]: """ Args: model: An instance of a supported data model (or dictionary of one) model_name: string name of the model passed (since model can be a dictionary) dsp_id: dsp identifier Returns: A dictionary of an image object or ``None`` """ route_group = cls._get_route_group(model_name) route_map = cls._get_route_map(route_group, dsp_id) resource_id = cls._get_model_value(model, route_map.id_field) remote_uri = cls._get_remote_uri(route_map, resource_id) if not (resource_id and remote_uri): return None data = AtlasImageModel(uri=remote_uri) return AtlasImageSchema().load(asdict(data)) @classmethod def _get_route_group(cls, model_name: str) -> Optional[AtlasRouteGroup]: """Determines :class:`AtlasRouteGroup` based on ``model_name`` if one exists.""" return MODEL_NAME_ENUM_MAP.get(model_name) @classmethod def _get_route_map(cls, route_group: AtlasRouteGroup, dsp_id: str = None) -> Optional[RouteMap]: """Determines :class:`RouteMap` based on :class:`AtlasRouteGroup` and ``dsp_id``.""" if not route_group: return None route_map = ATLAS_ROUTE_MAP.get(route_group) if route_group == AtlasRouteGroup.ARTIST: return route_map elif route_group == AtlasRouteGroup.PLAYLIST: if not dsp_id: LOG.error('Missing `dsp_id` for determining Atlas Playlist route') return None return route_map.get(dsp_id) else: return route_map @classmethod def _get_model_value(cls, model: Union[BaseModel, dict], field: str) -> Optional[str]: """Gets an attribute from an object, or a value by key from a dictionary""" if isinstance(model, dict): return model.get(field) try: return getattr(model, field, None) except AttributeError as e: LOG.exception('Unable to get attribute from an unexpected model type', error_details=str(e)) return None @classmethod def _get_remote_uri(cls, route_map: RouteMap, identifier) -> str: """ Returns: the full remote URI for the request to Atlas """ return urljoin(ATLAS_HOST, f'{route_map.route}/{identifier}')