""" LegacyImageAsset Model. LegacyImageAsset model uses sqlalchemy. It's used to store information about processed images. """ import datetime from typing import Any from sqlalchemy import Column, Integer, String from assets import config from assets.connectors.mysql import ArModel, ar_db_session from assets.constants import legacy_image_asset from assets.exceptions import ImageAssetNotFound class LegacyImageAsset(ArModel): """Table definition for `image_assets` table.""" __tablename__ = "image_assets" image_asset_id = Column("id", Integer, primary_key=True, autoincrement=True) category_id = Column(Integer) filename = Column(String(100), nullable=False) path = Column(String(100), nullable=False) height = Column(Integer, nullable=False) width = Column(Integer, nullable=False) mime_type = Column(String(50), nullable=False) file_size = Column(Integer, nullable=False) date_added = Column(String(50), default=datetime.datetime.now) date_modified = Column(String(50), default=datetime.datetime.now) cdn_url = Column(String(255)) def to_dict(self) -> dict[str, Any]: """Cast db record to dict. Returns: dict: Dictionary representation of object """ legacy_image_asset_dict = { "id": self.image_asset_id, "category_id": self.category_id, "path": self.path, "filename": self.filename, "height": self.height, "width": self.width, "mime_type": self.mime_type, "file_size": self.file_size, "cdn_url": self.cdn_url, } return legacy_image_asset_dict def get_legacy_image_url(asset: dict[str, Any]) -> str: """Given an asset get its CDN url. Args: asset (dict): representation of a legacy image asset row as a dict. Returns: (str): legacy image url. """ category_id = asset["category_id"] path = asset["path"][1:] category_path = legacy_image_asset.CATEGORY_ID_TO_PATH.get(category_id) if config.ENVIRONMENT == "prod": return legacy_image_asset.IMAGES_URL.format( category_path=category_path, path=path, filename=asset["filename"] ) return legacy_image_asset.IMAGES_QA_URL.format( category_path=category_path, path=path, filename=asset["filename"] ) def get_image_asset_url_by_id(image_asset_id: int) -> str: """Select an image_asset record by `image_asset_id` and return its url. Args: image_asset_id (int): unique identifier for an asset. Returns: str: image url """ with ar_db_session(read_only=True) as session: image_assets_result = ( session.query(LegacyImageAsset) .filter(LegacyImageAsset.image_asset_id == image_asset_id) .one_or_none() ) if not image_assets_result: raise ImageAssetNotFound() return get_legacy_image_url(image_assets_result.to_dict())