from __future__ import annotations from datetime import datetime from sqlalchemy import DateTime, ForeignKey, Index, Integer, String, UniqueConstraint, func from sqlalchemy.orm import Mapped, mapped_column from app.core.database import Base from app.utils.ids import prefixed_id class UsageOperation(Base): """One privacy-safe, idempotent reservation for an expensive operation.""" __tablename__ = "usage_operations" __table_args__ = ( UniqueConstraint( "user_id", "operation", "idempotency_key", name="uq_usage_operations_user_operation_key", ), Index( "ix_usage_operations_user_period_operation", "user_id", "period_key", "operation", ), ) id: Mapped[str] = mapped_column( String(40), primary_key=True, default=lambda: prefixed_id("usage") ) user_id: Mapped[str] = mapped_column( String(40), ForeignKey("users.id", ondelete="CASCADE"), nullable=False, index=True ) operation: Mapped[str] = mapped_column(String(48), nullable=False) period_key: Mapped[str] = mapped_column(String(16), nullable=False) idempotency_key: Mapped[str] = mapped_column(String(180), nullable=False) status: Mapped[str] = mapped_column(String(20), nullable=False, default="reserved") units: Mapped[int] = mapped_column(Integer, nullable=False, default=1) request_id: Mapped[str | None] = mapped_column(String(100), nullable=True) resource_id: Mapped[str | None] = mapped_column(String(80), nullable=True) created_at: Mapped[datetime] = mapped_column( DateTime(timezone=True), server_default=func.now(), nullable=False ) updated_at: Mapped[datetime] = mapped_column( DateTime(timezone=True), server_default=func.now(), onupdate=func.now(), nullable=False ) completed_at: Mapped[datetime | None] = mapped_column(DateTime(timezone=True), nullable=True)