"""Activity timeline ORM model and schemas.""" from __future__ import annotations from datetime import datetime from enum import StrEnum from typing import Any from pydantic import BaseModel, ConfigDict, Field from sqlalchemy import DateTime, Enum as SqlEnum, ForeignKey, Integer, func, text from sqlalchemy.dialects.postgresql import JSONB from sqlalchemy.orm import Mapped, mapped_column, relationship from app.models.base import Base class ActivityType(StrEnum): COMMENT = "comment" STATUS_CHANGED = "status_changed" TASK_CREATED = "task_created" SYSTEM = "system" class Activity(Base): """Represents a timeline event for a deal.""" __tablename__ = "activities" id: Mapped[int] = mapped_column(Integer, primary_key=True) deal_id: Mapped[int] = mapped_column(ForeignKey("deals.id", ondelete="CASCADE")) author_id: Mapped[int | None] = mapped_column( ForeignKey("users.id", ondelete="SET NULL"), nullable=True ) type: Mapped[ActivityType] = mapped_column(SqlEnum(ActivityType, name="activity_type"), nullable=False) payload: Mapped[dict[str, Any]] = mapped_column( JSONB, nullable=False, server_default=text("'{}'::jsonb"), ) created_at: Mapped[datetime] = mapped_column( DateTime(timezone=True), server_default=func.now(), nullable=False ) deal = relationship("Deal", back_populates="activities") author = relationship("User", back_populates="activities") class ActivityBase(BaseModel): deal_id: int author_id: int | None = None type: ActivityType payload: dict[str, Any] = Field(default_factory=dict) class ActivityCreate(ActivityBase): pass class ActivityRead(ActivityBase): id: int created_at: datetime model_config = ConfigDict(from_attributes=True)