import datetime from typing import Any import sqlalchemy as sa from fansifter_common.utils import timezone from fansifter_common.utils.uuid import uuid_string from sqlalchemy.orm import Mapped, mapped_column, reconstructor, relationship from preference_center.adapters.db import Model from preference_center.adapters.db.types import EnumType from preference_center.profile.enums import Gender from preference_center.profile.types import ProfileValues, ValueChange class TrackChangesMixin: @reconstructor def _init_state(self) -> None: self._origin_state = self.__dict__.copy() self._changed_fields: dict[str, ValueChange] = {} self._track_changes = True def __setattr__(self, name: str, value: Any) -> None: if hasattr(self, "_origin_state") and getattr(self, "_track_changes", True): if name in self._origin_state: old_value = self._origin_state[name] if old_value != value: self._changed_fields[name] = ValueChange( old_value=old_value, new_value=value ) super().__setattr__(name, value) def set_values(self, values: ProfileValues) -> None: """Update a profile.""" for field_name, value in values.items(): if hasattr(self, field_name): setattr(self, field_name, value) @property def has_changed(self) -> bool: """Check if the profile has changed.""" return bool(self._changed_fields) @property def changed_fields(self) -> dict[str, ValueChange]: """Get the changed fields.""" return self._changed_fields class Profile(TrackChangesMixin, Model, kw_only=True): __tablename__ = "fan_profile" id: Mapped[str] = mapped_column(primary_key=True, default_factory=uuid_string) email: Mapped[str] crm_id: Mapped[str | None] phone_number: Mapped[str | None] first_name: Mapped[str | None] last_name: Mapped[str | None] date_of_birth: Mapped[datetime.date | None] birthday: Mapped[str | None] gender: Mapped[Gender | None] = mapped_column( EnumType(Gender, processor=Gender.from_string) ) country_code: Mapped[str | None] city: Mapped[str | None] address: Mapped[str | None] zip_code: Mapped[str | None] updated_at: Mapped[datetime.datetime] = mapped_column( default_factory=timezone.now, onupdate=timezone.now ) deleted: Mapped[bool] = mapped_column(default=False) updated_by: Mapped[str | None] def __post_init__(self) -> None: # Set the birthday if the date of birth is set if self.date_of_birth and not self.birthday: self.birthday = self.date_of_birth.strftime("%m/%d") # Initialize the state self._init_state() class MailingList(Model, kw_only=True): __tablename__ = "fan_mailing_list" id: Mapped[str] = mapped_column(primary_key=True, default_factory=uuid_string) name: Mapped[str] class Subscription(Model, kw_only=True): __tablename__ = "fan_subscription" id: Mapped[str] = mapped_column(primary_key=True, default_factory=uuid_string) profile_id: Mapped[str] = mapped_column(sa.ForeignKey("fan_profile.id")) mailing_list_id: Mapped[str] = mapped_column( sa.ForeignKey("fan_mailing_list.id"), init=False ) is_active: Mapped[bool] updated_at: Mapped[datetime.datetime] = mapped_column( default_factory=timezone.now, onupdate=timezone.now ) mailing_list: Mapped[MailingList] = relationship(lazy="raise") def __post_init__(self) -> None: self.mailing_list_id = self.mailing_list.id @property def name(self) -> str: return self.mailing_list.name