from collections.abc import Mapping from datetime import datetime from typing import Any import sqlalchemy as sa from sqlalchemy.orm import Mapped, declared_attr, mapped_column, relationship from campaigns.assets.enums import AssetStatus from campaigns.assets.models import Asset from campaigns.assets.types import AssetError class AssetMixin: asset_id: Mapped[str] = mapped_column(sa.ForeignKey("asset.id")) _errors: Mapped[list[Mapping[str, Any]] | None] = mapped_column( sa.JSON, name="errors" ) is_valid: Mapped[bool | None] = mapped_column(default=None) @declared_attr def asset(self) -> Mapped[Asset]: return relationship("Asset", lazy="raise") @property def filename(self) -> str: return self.asset.filename @property def url(self) -> str | None: return self.asset.url @property def thumbnail_url(self) -> str | None: return self.asset.thumbnail_url @property def size(self) -> int | None: return self.asset.size @property def uploaded_at(self) -> datetime | None: return self.asset.uploaded_at @property def errors(self) -> list[Mapping[str, Any]]: return [*(self.asset.errors or []), *(self._errors or [])] def add_error(self, actual: str, expected: list[str], code: str) -> None: if self._errors is None: self._errors = [] self._errors.append(AssetError(actual=actual, expected=expected, code=code)) @property def status(self) -> AssetStatus: if self.asset.status == AssetStatus.INVALID or self.is_valid is False: return AssetStatus.INVALID elif self.is_valid is None: return AssetStatus.PROCESSING return AssetStatus.COMPLETED