"""Logic for Image Location.""" from assets import config from assets.constants import ( asset_types as asset_types_constants, error, s3, ) from assets.exceptions import ImageAssetNotFound from assets.models import image_location as image_location_model, ows_product, s3_file from assets.models.legacy import image_asset def get_image_location(entity_id: int, image_type: str, image_format: str) -> str: """Get image location. Gets location for image or returns error response if it doesn't exist in s3 bucket. Args: entity_id (int): Unique identifier for the product or artist or vendor. image_type: (str): Type of entity (e.g. "product" or "artist" or "vendor"). image_format: (str): Whether this is for a thumbnail or cover. Returns: str: the path for the image. """ if ( image_format == asset_types_constants.SUBTYPE_XLARGE_COVER and image_type == "product" ): product = ows_product.get_product_by_id(product_id=entity_id) filename = "{}.jpg".format(product["upc"]) cdn = config.XLARGE_COVER_CDN_URL bucket = s3.XLARGE_COVER_BUCKET else: filename = image_location_model.get_image_filename( entity_id=entity_id, image_type=image_type, image_format=image_format ) cdn = config.CDN_URL bucket = s3.IMAGE_ASSET_BUCKET if not _check_if_file_on_s3(filename, bucket=bucket): raise ImageAssetNotFound(error.ERROR_IMAGE_NOT_FOUND) return image_location_model.get_image_location_by_filename( filename=filename, cdn=cdn ) def get_image_locations( entity_ids: list[int], image_type: str, image_format: str ) -> dict[int, str]: """Get image locations. Gets locations for images or returns error response if an image doesn't exist in the s3 bucket. Note that this function doesn't support x-large image formats, and that it doesn't check existence of underlying assets on S3. It assumes the client handles image URLs of unlocated assets Args: entity_ids (list[int]): List of product or artist identifiers. image_type: (str): Type of entity (e.g. "product" or "artist"). image_format: (str): Whether this is for a thumbnail or cover. Returns: dict: Paths for images. """ return image_location_model.get_v1_images_locations( entity_ids=entity_ids, image_type=image_type, image_format=image_format ) def _check_if_file_on_s3(filename: str, bucket: str) -> bool: """Check file exists on S3 in given bucket.""" file_path = "{bucket}/{filename}".format(bucket=bucket, filename=filename) return s3_file.check_s3_file_exists(config.ASSET_STORAGE_BUCKET_NAME, file_path) def get_vendor_icon(vendor_id: int, asset_id: int) -> str: """Get image location. Gets location for image first by S3 and then falls back to legacy image retrieval if not found. Args: vendor_id (int): Unique identifier for the vendor. asset_id (int): Unique identifier for an image_asset. Returns: str: the path for the image. """ try: return get_image_location(vendor_id, s3.VENDOR_ENTITY, s3.VENDOR_ICON_FOLDER) except ImageAssetNotFound: return image_asset.get_image_asset_url_by_id(asset_id)