import typing from apollo_utils.core.constants.dsp import DSP from marshmallow import fields from marshmallow.validate import OneOf from typing import List from server.constants.core_image_service import CORE_IMAGE_SERVICE_ALLOWED_RESOLUTIONS, \ CORE_IMAGE_SERVICE_UNDESIRABLE_IMAGES_DSP_ENTITY_PAIRS from server.utils.images.core_image_service import core_image_service_add_resolution def get_core_image_service_image_size_mixin( data_key: str = "image_size", load_default: int = 640, allowed_resolutions: List[int] = CORE_IMAGE_SERVICE_ALLOWED_RESOLUTIONS, *args, **kwargs, ): class ImageSizeMixin: image_size = fields.Integer( data_key=data_key, validate=OneOf(allowed_resolutions), load_default=load_default, *args, **kwargs ) return ImageSizeMixin class CoreImageServiceImageUrlField(fields.String): def __init__(self, add_resolution: bool = True, default_resolution: int = 640, *args, **kwargs): self._add_resolution = add_resolution self._default_resolution = default_resolution super().__init__(*args, **kwargs) def _serialize(self, value, attr, obj, **kwargs) -> typing.Optional[str]: field_value = super()._serialize(value, attr, obj, **kwargs) if self._add_resolution: field_value = core_image_service_add_resolution( field_value, self.context.get("image_size", self._default_resolution) ) return field_value CoreImageServiceImageSizeMixin = get_core_image_service_image_size_mixin() def get_core_image_service_availability(dsp: DSP, entity: str) -> bool: """Check whether Core Image Service image_url is available by DSP and entity Args: dsp: DSP entity: One of ["track", "playlist"] string values Returns: Boolean """ return entity not in CORE_IMAGE_SERVICE_UNDESIRABLE_IMAGES_DSP_ENTITY_PAIRS.get(dsp.value, [])