73 lines
1.9 KiB
Python
73 lines
1.9 KiB
Python
"""Event model for audit logging."""
|
|
|
|
import uuid
|
|
from datetime import datetime, timezone
|
|
from typing import TYPE_CHECKING, Any
|
|
|
|
from sqlalchemy import DateTime, ForeignKey, JSON, String
|
|
from sqlalchemy.dialects.postgresql import JSONB
|
|
from sqlalchemy.orm import Mapped, mapped_column, relationship
|
|
|
|
# Use JSONB on PostgreSQL, JSON on other databases (SQLite for tests)
|
|
JSONType = JSON().with_variant(JSONB, "postgresql")
|
|
|
|
from app.models.base import Base, UUIDMixin
|
|
|
|
if TYPE_CHECKING:
|
|
from app.models.task import Task
|
|
from app.models.tenant import Tenant
|
|
|
|
|
|
def utc_now() -> datetime:
|
|
"""Return current UTC datetime."""
|
|
return datetime.now(timezone.utc)
|
|
|
|
|
|
class Event(UUIDMixin, Base):
|
|
"""
|
|
Event model for audit logging and activity tracking.
|
|
|
|
Events are immutable records of system activity.
|
|
Only has created_at (no updated_at since events are immutable).
|
|
"""
|
|
|
|
__tablename__ = "events"
|
|
|
|
tenant_id: Mapped[uuid.UUID] = mapped_column(
|
|
ForeignKey("tenants.id", ondelete="CASCADE"),
|
|
nullable=False,
|
|
index=True,
|
|
)
|
|
task_id: Mapped[uuid.UUID | None] = mapped_column(
|
|
ForeignKey("tasks.id", ondelete="SET NULL"),
|
|
nullable=True,
|
|
index=True,
|
|
)
|
|
event_type: Mapped[str] = mapped_column(
|
|
String(100),
|
|
nullable=False,
|
|
index=True,
|
|
)
|
|
payload: Mapped[dict[str, Any]] = mapped_column(
|
|
JSONType,
|
|
nullable=False,
|
|
default=dict,
|
|
)
|
|
created_at: Mapped[datetime] = mapped_column(
|
|
DateTime(timezone=True),
|
|
default=utc_now,
|
|
nullable=False,
|
|
index=True,
|
|
)
|
|
|
|
# Relationships
|
|
tenant: Mapped["Tenant"] = relationship(
|
|
back_populates="events",
|
|
)
|
|
task: Mapped["Task | None"] = relationship(
|
|
back_populates="events",
|
|
)
|
|
|
|
def __repr__(self) -> str:
|
|
return f"<Event(id={self.id}, type={self.event_type})>"
|