63 lines
1.8 KiB
Python
63 lines
1.8 KiB
Python
"""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)
|