from __future__ import annotations import enum import uuid from datetime import datetime from sqlalchemy import ( DateTime, Enum, ForeignKey, Integer, String, UniqueConstraint, func, ) from sqlalchemy.dialects.postgresql import UUID from sqlalchemy.orm import Mapped, mapped_column, relationship from app.database.database import Base from typing import TYPE_CHECKING if TYPE_CHECKING: from app.models.document_chunk import DocumentChunk class DocumentVersionStatus(str, enum.Enum): UPLOADED = "UPLOADED" PROCESSING = "PROCESSING" PROCESSED = "PROCESSED" FAILED = "FAILED" class DocumentVersion(Base): """ Represents a specific uploaded version of a document. Each upload creates a new DocumentVersion while preserving the logical Document identity. """ __tablename__ = "document_versions" __table_args__ = ( UniqueConstraint( "document_id", "version_number", name="uq_document_version", ), ) id: Mapped[uuid.UUID] = mapped_column( UUID(as_uuid=True), primary_key=True, default=uuid.uuid4, ) document_id: Mapped[uuid.UUID] = mapped_column( UUID(as_uuid=True), ForeignKey("documents.id", ondelete="CASCADE"), nullable=False, index=True, ) version_number: Mapped[int] = mapped_column( Integer, nullable=False, index=True, ) filename: Mapped[str] = mapped_column( String(255), nullable=False, ) file_type: Mapped[str] = mapped_column( String(20), nullable=False, ) checksum: Mapped[str] = mapped_column( String(128), nullable=False, index=True, ) storage_path: Mapped[str] = mapped_column( String(1000), nullable=False, ) status: Mapped[DocumentVersionStatus] = mapped_column( Enum(DocumentVersionStatus, name="document_version_status"), nullable=False, default=DocumentVersionStatus.UPLOADED, server_default=DocumentVersionStatus.UPLOADED.value, index=True, ) uploaded_by: Mapped[uuid.UUID] = mapped_column( UUID(as_uuid=True), ForeignKey("users.id"), nullable=False, ) uploaded_at: Mapped[datetime] = mapped_column( DateTime(timezone=True), nullable=False, server_default=func.now(), ) processed_at: Mapped[datetime | None] = mapped_column( DateTime(timezone=True), nullable=True, ) # ------------------------------------------------------------------ # Relationships # ------------------------------------------------------------------ document: Mapped["Document"] = relationship( back_populates="versions", ) uploader: Mapped["User"] = relationship() knowledge_items: Mapped[list["KnowledgeItem"]] = relationship( back_populates="document_version", cascade="all, delete-orphan", passive_deletes=True, ) evidence: Mapped[list["KnowledgeEvidence"]] = relationship( back_populates="document_version", cascade="all, delete-orphan", passive_deletes=True, ) chunks: Mapped[list["DocumentChunk"]] = relationship( back_populates="document_version", cascade="all, delete-orphan", passive_deletes=True, ) workflow_runs: Mapped[list["WorkflowRun"]] = relationship( back_populates="document_version", cascade="all, delete-orphan", passive_deletes=True, ) def __repr__(self) -> str: return ( f"" )