letsbe-orchestrator/app/models/event.py

70 lines
1.8 KiB
Python
Raw Normal View History

"""Event model for audit logging."""
import uuid
from datetime import datetime, timezone
from typing import TYPE_CHECKING, Any
from sqlalchemy import DateTime, ForeignKey, String
from sqlalchemy.dialects.postgresql import JSONB
from sqlalchemy.orm import Mapped, mapped_column, relationship
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(
JSONB,
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})>"