from __future__ import annotations import urllib.parse from datetime import datetime from typing import TYPE_CHECKING, Any, ClassVar import sqlalchemy as sa from audience_common import datetimeutc from sqlalchemy.orm import Mapped, mapped_column, relationship from campaigns.config import settings from campaigns.connectors.db.orm import Base from campaigns.connectors.db.sqltypes import EncryptedTextType from campaigns.connectors.facebook import models as fb_models from campaigns.core.types import EncryptedToken, PlainToken from campaigns.utils.uuid import uuid_string if TYPE_CHECKING: from .facebook_association import FacebookAssociation from .facebook_page_label import FacebookPageLabel class FacebookPage(Base): __tablename__ = "facebook_page" __table_args__ = (sa.UniqueConstraint("association_id", "external_id"),) UPLOAD_FOLDER: ClassVar[str] = "facebook-pages" id: Mapped[str] = mapped_column(primary_key=True, default=uuid_string) association_id: Mapped[str] = mapped_column( sa.ForeignKey("facebook_association.id") ) external_id: Mapped[str] name: Mapped[str] token: Mapped[EncryptedToken] = mapped_column(EncryptedTextType) is_manageable: Mapped[bool] is_valid: Mapped[bool] = mapped_column(default=True) is_agency_provider: Mapped[bool] = mapped_column(default=False) picture_key: Mapped[str | None] = mapped_column(unique=True) picture_bucket: Mapped[str | None] created_at: Mapped[datetime] = mapped_column(default=datetimeutc.now) updated_at: Mapped[datetime] = mapped_column( default=datetimeutc.now, onupdate=datetimeutc.now ) # relations association: Mapped[FacebookAssociation] = relationship( back_populates="pages", lazy="raise" ) labels: Mapped[list[FacebookPageLabel]] = relationship( back_populates="page", lazy="raise" ) def __init__(self, **kw: Any) -> None: super().__init__(**kw) self._plain_token: PlainToken | None = None @classmethod def from_api_obj(cls, api_obj: fb_models.Page) -> FacebookPage: return cls( external_id=api_obj.id, name=api_obj.name, is_manageable=api_obj.is_manageable, ) @property def plain_token(self) -> PlainToken: if self._plain_token is None: raise AttributeError("Plain token is not set.") return self._plain_token @plain_token.setter def plain_token(self, val: PlainToken) -> None: self._plain_token = val @property def picture_url(self) -> str | None: if self.picture_key: return urllib.parse.urljoin( f"https://{settings.assets_cdn_domain}/", urllib.parse.quote_plus(self.picture_key, safe="/"), ) else: return None def set_picture_bucket_and_key(self) -> None: self.picture_bucket = settings.assets_bucket_name self.picture_key = f"{self.UPLOAD_FOLDER}/{self.id}/picture.png"