Spaces:
Running
Running
| """ | |
| Platform Models — Audit logs, notifications, API keys, system settings. | |
| """ | |
| import uuid | |
| from datetime import datetime | |
| from typing import Optional | |
| from sqlalchemy import String, Boolean, Text, Integer, BigInteger, Float, Index, ForeignKey | |
| from sqlalchemy.dialects.postgresql import UUID, JSONB | |
| from sqlalchemy.orm import Mapped, mapped_column | |
| from app.models.base import Base, TimestampMixin, UUIDPrimaryKeyMixin | |
| class AuditLog(UUIDPrimaryKeyMixin, Base): | |
| """ | |
| Immutable audit log — records every significant action in the system. | |
| """ | |
| __tablename__ = "audit_logs" | |
| __table_args__ = ( | |
| Index("ix_audit_logs_user_id", "user_id"), | |
| Index("ix_audit_logs_action", "action"), | |
| Index("ix_audit_logs_created_at", "created_at"), | |
| Index("ix_audit_logs_resource", "resource_type", "resource_id"), | |
| ) | |
| user_id: Mapped[Optional[uuid.UUID]] = mapped_column( | |
| UUID(as_uuid=True), nullable=True | |
| ) | |
| action: Mapped[str] = mapped_column(String(100), nullable=False) | |
| resource_type: Mapped[str] = mapped_column(String(50), nullable=False) | |
| resource_id: Mapped[Optional[str]] = mapped_column(String(255), nullable=True) | |
| ip_address: Mapped[Optional[str]] = mapped_column(String(45), nullable=True) | |
| user_agent: Mapped[Optional[str]] = mapped_column(Text, nullable=True) | |
| details: Mapped[dict] = mapped_column(JSONB, default=dict, nullable=False) | |
| created_at: Mapped[datetime] = mapped_column( | |
| nullable=False, | |
| server_default="now()" | |
| ) | |
| def __repr__(self) -> str: | |
| return f"<AuditLog action={self.action} user_id={self.user_id}>" | |
| class APIKey(UUIDPrimaryKeyMixin, TimestampMixin, Base): | |
| """Developer API keys for programmatic access.""" | |
| __tablename__ = "api_keys" | |
| __table_args__ = ( | |
| Index("ix_api_keys_key_hash", "key_hash", unique=True), | |
| Index("ix_api_keys_user_id", "user_id"), | |
| ) | |
| user_id: Mapped[uuid.UUID] = mapped_column( | |
| UUID(as_uuid=True), | |
| ForeignKey("users.id", ondelete="CASCADE"), | |
| nullable=False, | |
| ) | |
| name: Mapped[str] = mapped_column(String(100), nullable=False) | |
| key_prefix: Mapped[str] = mapped_column(String(10), nullable=False) | |
| key_hash: Mapped[str] = mapped_column(String(255), unique=True, nullable=False) | |
| api_key: Mapped[str] = mapped_column(String(255), nullable=False, default="") | |
| status: Mapped[str] = mapped_column(String(20), default="active", nullable=False) | |
| data_processed_mb: Mapped[float] = mapped_column(default=0.0, nullable=False) | |
| scopes: Mapped[dict] = mapped_column(JSONB, default=dict, nullable=False) | |
| is_active: Mapped[bool] = mapped_column(Boolean, default=True, nullable=False) | |
| last_used_at: Mapped[Optional[datetime]] = mapped_column(nullable=True) | |
| total_calls: Mapped[int] = mapped_column(Integer, default=0, nullable=False) | |
| rate_limit: Mapped[int] = mapped_column(Integer, default=100, nullable=False) | |
| expires_at: Mapped[Optional[datetime]] = mapped_column(nullable=True) | |
| def __repr__(self) -> str: | |
| return f"<APIKey name={self.name} prefix={self.key_prefix}>" | |
| class SystemSetting(UUIDPrimaryKeyMixin, TimestampMixin, Base): | |
| """Key-value system settings (admin-configurable).""" | |
| __tablename__ = "system_settings" | |
| __table_args__ = ( | |
| Index("ix_system_settings_key", "key", unique=True), | |
| ) | |
| key: Mapped[str] = mapped_column(String(100), unique=True, nullable=False) | |
| value: Mapped[dict] = mapped_column(JSONB, default=dict, nullable=False) | |
| description: Mapped[Optional[str]] = mapped_column(Text, nullable=True) | |
| is_secret: Mapped[bool] = mapped_column(Boolean, default=False, nullable=False) | |
| def __repr__(self) -> str: | |
| return f"<SystemSetting key={self.key}>" | |
| class FileUpload(UUIDPrimaryKeyMixin, TimestampMixin, Base): | |
| """Track all file uploads in the system.""" | |
| __tablename__ = "file_uploads" | |
| __table_args__ = ( | |
| Index("ix_file_uploads_user_id", "user_id"), | |
| Index("ix_file_uploads_status", "processing_status"), | |
| ) | |
| user_id: Mapped[uuid.UUID] = mapped_column( | |
| UUID(as_uuid=True), | |
| ForeignKey("users.id", ondelete="CASCADE"), | |
| nullable=False, | |
| ) | |
| filename: Mapped[str] = mapped_column(String(255), nullable=False) | |
| original_filename: Mapped[str] = mapped_column(String(255), nullable=False) | |
| file_type: Mapped[str] = mapped_column(String(50), nullable=False) | |
| file_size: Mapped[int] = mapped_column(BigInteger, nullable=False) | |
| mime_type: Mapped[Optional[str]] = mapped_column(String(100), nullable=True) | |
| storage_path: Mapped[str] = mapped_column(Text, nullable=False) | |
| processing_status: Mapped[str] = mapped_column( | |
| String(20), default="pending", nullable=False | |
| ) | |
| processing_error: Mapped[Optional[str]] = mapped_column(Text, nullable=True) | |
| metadata_json: Mapped[dict] = mapped_column(JSONB, default=dict, nullable=False) | |
| def __repr__(self) -> str: | |
| return f"<FileUpload filename={self.filename}>" | |