generated from john/python-template
441 lines
18 KiB
Python
441 lines
18 KiB
Python
import logging
|
|
import shutil
|
|
from collections.abc import Sequence
|
|
from dataclasses import dataclass
|
|
from datetime import UTC
|
|
from datetime import datetime
|
|
from uuid import UUID
|
|
|
|
from sqlalchemy import func
|
|
from sqlalchemy.exc import IntegrityError
|
|
from sqlalchemy.orm import selectinload
|
|
from sqlmodel import col
|
|
from sqlmodel import select
|
|
from sqlmodel.ext.asyncio.session import AsyncSession
|
|
|
|
from ..db.models import Document
|
|
from ..db.models import DocumentPerson
|
|
from ..db.models import DocumentType
|
|
from ..errors import AppError
|
|
from ..errors import ErrorCategory
|
|
from .base import ServiceBase
|
|
|
|
logger = logging.getLogger(__name__)
|
|
|
|
|
|
class DocumentError(AppError):
|
|
"""Raised when document operations fail."""
|
|
|
|
|
|
class MissingSourceError(DocumentError):
|
|
"""Raised when a document has no associated sources."""
|
|
|
|
|
|
class DocumentAlreadyExistsError(DocumentError):
|
|
"""Raised when a document with the same name already exists in the database."""
|
|
|
|
|
|
class DocumentDeleteBlockedError(DocumentError):
|
|
"""Raised when a document delete is blocked by dependent records."""
|
|
|
|
|
|
class DocumentTypeError(DocumentError):
|
|
"""Raised when Document Type maintenance fails."""
|
|
|
|
|
|
def _normalize_registry_label(label: str) -> str:
|
|
normalized = label.strip()
|
|
if not normalized:
|
|
raise DocumentTypeError(
|
|
"Document Type label is required",
|
|
category=ErrorCategory.VALIDATION,
|
|
suggestion="Enter a user-facing label and retry.",
|
|
)
|
|
return normalized
|
|
|
|
|
|
def _document_type_label_key(label: str) -> str:
|
|
return _normalize_registry_label(label).casefold()
|
|
|
|
|
|
@dataclass(frozen=True, slots=True)
|
|
class DocumentTypeSummary:
|
|
"""Settings read model for a Document Type and its usage count."""
|
|
|
|
id: UUID
|
|
label: str
|
|
is_active: bool
|
|
document_count: int
|
|
|
|
|
|
class DocumentService(ServiceBase):
|
|
"""Thin service class for managing documents in the database."""
|
|
|
|
async def _validate_document_type(self, *, session: AsyncSession, document: Document) -> None:
|
|
"""Validate the UUID-backed Document Type reference."""
|
|
if document.document_type_id is None:
|
|
return
|
|
if await session.get(DocumentType, document.document_type_id) is None:
|
|
raise DocumentError(
|
|
f"Document type with id {document.document_type_id} not found",
|
|
category=ErrorCategory.VALIDATION,
|
|
suggestion="Select a valid document type and retry.",
|
|
)
|
|
|
|
async def _get_document_or_raise(self, *, session: AsyncSession, document_id: UUID) -> Document:
|
|
"""Get a document by id or raise a not-found service error."""
|
|
document = await session.get(Document, document_id)
|
|
if document is None:
|
|
raise DocumentError(
|
|
f"Document with id {document_id} not found",
|
|
category=ErrorCategory.NOT_FOUND,
|
|
suggestion="Verify the document id and retry.",
|
|
)
|
|
return document
|
|
|
|
#
|
|
# CRUD Operations
|
|
#
|
|
|
|
async def create_document(
|
|
self,
|
|
document: Document,
|
|
*,
|
|
session: AsyncSession | None = None,
|
|
) -> Document:
|
|
"""Create a new document in the database."""
|
|
async with self._session_scope(session) as _session:
|
|
await self._validate_document_type(session=_session, document=document)
|
|
_session.add(document)
|
|
try:
|
|
await self._finalize(session=_session, caller_session=session, refresh=(document,))
|
|
except IntegrityError as exc:
|
|
raise DocumentAlreadyExistsError(
|
|
f"Document with id {document.id} already exists",
|
|
category=ErrorCategory.VALIDATION,
|
|
suggestion="Rename the file and try again.",
|
|
) from exc
|
|
return document
|
|
|
|
async def read_document(self, document_id: UUID, *, session: AsyncSession | None = None) -> Document:
|
|
"""Read an existing document from the database.
|
|
|
|
The selectinload option is used to eagerly load related jobs and sources.
|
|
"""
|
|
async with self._session_scope(session) as _session:
|
|
document = await _session.get(
|
|
Document,
|
|
document_id,
|
|
options=(
|
|
selectinload(Document.jobs), # pyright: ignore[reportArgumentType]
|
|
selectinload(Document.sources), # pyright: ignore[reportArgumentType]
|
|
),
|
|
)
|
|
if document is None:
|
|
raise DocumentError(
|
|
f"Document with id {document_id} not found",
|
|
category=ErrorCategory.NOT_FOUND,
|
|
suggestion="Re-upload the source document and retry.",
|
|
)
|
|
elif not document.sources:
|
|
raise MissingSourceError(
|
|
f"Document with id {document_id} has no associated source records",
|
|
category=ErrorCategory.NOT_FOUND,
|
|
suggestion="Upload at least one source for this document and retry.",
|
|
)
|
|
return document
|
|
|
|
async def update_document(self, document: Document, *, session: AsyncSession | None = None) -> Document:
|
|
"""Update an existing document in the database."""
|
|
async with self._session_scope(session) as _session:
|
|
await self._validate_document_type(session=_session, document=document)
|
|
document.updated_at = datetime.now(UTC)
|
|
merged = await _session.merge(document)
|
|
await self._finalize(session=_session, caller_session=session, refresh=(merged,))
|
|
return merged
|
|
|
|
async def delete_document(self, document: Document, *, session: AsyncSession | None = None) -> None:
|
|
"""Delete a document from the database."""
|
|
document_id = document.id
|
|
async with self._session_scope(session) as _session:
|
|
existing = await _session.get(
|
|
Document,
|
|
document.id,
|
|
options=(
|
|
selectinload(Document.jobs), # pyright: ignore[reportArgumentType]
|
|
selectinload(Document.sources), # pyright: ignore[reportArgumentType]
|
|
selectinload(Document.document_people), # pyright: ignore[reportArgumentType]
|
|
),
|
|
)
|
|
if existing is None:
|
|
raise DocumentError(
|
|
f"Document with id {document.id} not found",
|
|
category=ErrorCategory.NOT_FOUND,
|
|
suggestion="Verify the document id and retry.",
|
|
)
|
|
|
|
has_jobs = bool(existing.jobs)
|
|
has_sources = bool(existing.sources)
|
|
if has_jobs or has_sources:
|
|
blocked_by: list[str] = []
|
|
if has_sources:
|
|
blocked_by.append("Sources")
|
|
if has_jobs:
|
|
blocked_by.append("Jobs")
|
|
raise DocumentDeleteBlockedError(
|
|
f"Document delete blocked by related records: {', '.join(blocked_by)}",
|
|
category=ErrorCategory.VALIDATION,
|
|
suggestion="Remove related Sources and Jobs first, then retry deletion.",
|
|
)
|
|
|
|
for link in list(existing.document_people):
|
|
await _session.delete(link)
|
|
|
|
await _session.delete(existing)
|
|
await self._finalize(session=_session, caller_session=session)
|
|
|
|
self._delete_document_storage_folder(document_id=document_id)
|
|
|
|
def _delete_document_storage_folder(self, *, document_id: UUID) -> None:
|
|
"""Best-effort cleanup for document-scoped source storage."""
|
|
document_dir = self.settings.upload_dir / "documents" / str(document_id)
|
|
if not document_dir.exists():
|
|
return
|
|
|
|
try:
|
|
shutil.rmtree(document_dir)
|
|
logger.info("Deleted document storage folder: %s", document_dir)
|
|
except OSError:
|
|
logger.warning("Failed to delete document storage folder: %s", document_dir)
|
|
|
|
# Query Operations
|
|
|
|
async def query_documents(
|
|
self, *, name: str | None = None, session: AsyncSession | None = None
|
|
) -> Sequence[Document]:
|
|
"""Query documents from the database based on provided filters."""
|
|
async with self._session_scope(session) as _session:
|
|
query = select(Document)
|
|
if name is not None:
|
|
query = query.where(Document.name == name)
|
|
result = await _session.exec(query)
|
|
return result.all()
|
|
|
|
async def list_documents(self, *, session: AsyncSession | None = None) -> Sequence[Document]:
|
|
"""List documents with relations needed by the archival table."""
|
|
async with self._session_scope(session) as _session:
|
|
query = select(Document).options(
|
|
selectinload(Document.document_people).selectinload(DocumentPerson.person), # pyright: ignore[reportArgumentType]
|
|
selectinload(Document.document_people).selectinload(DocumentPerson.role_ref), # pyright: ignore[reportArgumentType]
|
|
selectinload(Document.document_type_ref), # pyright: ignore[reportArgumentType]
|
|
)
|
|
result = await _session.exec(query)
|
|
return result.all()
|
|
|
|
async def read_document_detail(self, document_id: UUID, *, session: AsyncSession | None = None) -> Document:
|
|
"""Read a document with eagerly loaded relations for UI detail rendering."""
|
|
async with self._session_scope(session) as _session:
|
|
query = (
|
|
select(Document)
|
|
.options(
|
|
selectinload(Document.jobs), # pyright: ignore[reportArgumentType]
|
|
selectinload(Document.sources), # pyright: ignore[reportArgumentType]
|
|
selectinload(Document.document_people).selectinload(DocumentPerson.person), # pyright: ignore[reportArgumentType]
|
|
selectinload(Document.document_people).selectinload(DocumentPerson.role_ref), # pyright: ignore[reportArgumentType]
|
|
selectinload(Document.document_type_ref), # pyright: ignore[reportArgumentType]
|
|
)
|
|
.where(Document.id == document_id)
|
|
.execution_options(populate_existing=True)
|
|
)
|
|
document = (await _session.exec(query)).first()
|
|
if document is None:
|
|
raise DocumentError(
|
|
f"Document with id {document_id} not found",
|
|
category=ErrorCategory.NOT_FOUND,
|
|
suggestion="Verify the document id and retry.",
|
|
)
|
|
return document
|
|
|
|
async def list_document_types(
|
|
self,
|
|
*,
|
|
active_only: bool = True,
|
|
session: AsyncSession | None = None,
|
|
) -> Sequence[DocumentType]:
|
|
"""List configured document types."""
|
|
async with self._session_scope(session) as _session:
|
|
query = select(DocumentType)
|
|
if active_only:
|
|
query = query.where(col(DocumentType.is_active).is_(True))
|
|
query = query.order_by(col(DocumentType.normalized_label), col(DocumentType.id))
|
|
result = await _session.exec(query)
|
|
return result.all()
|
|
|
|
async def list_document_type_summaries(
|
|
self,
|
|
*,
|
|
session: AsyncSession | None = None,
|
|
) -> Sequence[DocumentTypeSummary]:
|
|
"""List Document Types alphabetically with current usage counts."""
|
|
async with self._session_scope(session) as _session:
|
|
query = (
|
|
select(DocumentType, func.count(col(Document.id)))
|
|
.outerjoin(Document, col(Document.document_type_id) == col(DocumentType.id))
|
|
.group_by(col(DocumentType.id))
|
|
.order_by(col(DocumentType.normalized_label), col(DocumentType.id))
|
|
)
|
|
rows = (await _session.exec(query)).all()
|
|
return [
|
|
DocumentTypeSummary(
|
|
id=document_type.id,
|
|
label=document_type.label,
|
|
is_active=document_type.is_active,
|
|
document_count=int(document_count),
|
|
)
|
|
for document_type, document_count in rows
|
|
]
|
|
|
|
async def create_document_type(
|
|
self,
|
|
*,
|
|
label: str,
|
|
is_active: bool = True,
|
|
session: AsyncSession | None = None,
|
|
) -> DocumentType:
|
|
"""Create a UUID-identified Document Type with a unique label."""
|
|
document_type = DocumentType(
|
|
label=_normalize_registry_label(label),
|
|
normalized_label=_document_type_label_key(label),
|
|
is_active=is_active,
|
|
)
|
|
async with self._session_scope(session) as _session:
|
|
_session.add(document_type)
|
|
try:
|
|
await self._finalize(session=_session, caller_session=session, refresh=(document_type,))
|
|
except IntegrityError as exc:
|
|
raise DocumentTypeError(
|
|
f"Document Type label {document_type.label!r} already exists",
|
|
category=ErrorCategory.CONFLICT,
|
|
suggestion="Choose a different label or edit the existing type.",
|
|
) from exc
|
|
return document_type
|
|
|
|
async def read_document_type(
|
|
self,
|
|
document_type_id: UUID,
|
|
*,
|
|
session: AsyncSession | None = None,
|
|
) -> DocumentType:
|
|
"""Read a Document Type by id."""
|
|
async with self._session_scope(session) as _session:
|
|
document_type = await _session.get(DocumentType, document_type_id)
|
|
if document_type is None:
|
|
raise DocumentTypeError(
|
|
f"Document Type with id {document_type_id} not found",
|
|
category=ErrorCategory.NOT_FOUND,
|
|
suggestion="Refresh Settings and select an available Document Type.",
|
|
)
|
|
return document_type
|
|
|
|
async def update_document_type(
|
|
self,
|
|
document_type_id: UUID,
|
|
*,
|
|
label: str,
|
|
is_active: bool,
|
|
session: AsyncSession | None = None,
|
|
) -> DocumentType:
|
|
"""Update a Document Type label and active state."""
|
|
async with self._session_scope(session) as _session:
|
|
document_type = await _session.get(DocumentType, document_type_id)
|
|
if document_type is None:
|
|
raise DocumentTypeError(
|
|
f"Document Type with id {document_type_id} not found",
|
|
category=ErrorCategory.NOT_FOUND,
|
|
suggestion="Refresh Settings and select an available Document Type.",
|
|
)
|
|
document_type.label = _normalize_registry_label(label)
|
|
document_type.normalized_label = _document_type_label_key(label)
|
|
document_type.is_active = is_active
|
|
document_type.updated_at = datetime.now(UTC)
|
|
try:
|
|
await self._finalize(session=_session, caller_session=session, refresh=(document_type,))
|
|
except IntegrityError as exc:
|
|
raise DocumentTypeError(
|
|
f"Document Type label {document_type.label!r} already exists",
|
|
category=ErrorCategory.CONFLICT,
|
|
suggestion="Choose a different label or edit the existing type.",
|
|
) from exc
|
|
return document_type
|
|
|
|
async def delete_document_type(
|
|
self,
|
|
document_type_id: UUID,
|
|
*,
|
|
session: AsyncSession | None = None,
|
|
) -> None:
|
|
"""Delete an unreferenced Document Type without cascade behavior."""
|
|
async with self._session_scope(session) as _session:
|
|
document_type = await _session.get(DocumentType, document_type_id)
|
|
if document_type is None:
|
|
raise DocumentTypeError(
|
|
f"Document Type with id {document_type_id} not found",
|
|
category=ErrorCategory.NOT_FOUND,
|
|
suggestion="Refresh Settings and select an available Document Type.",
|
|
)
|
|
if await self._document_type_is_referenced(session=_session, document_type=document_type):
|
|
raise DocumentTypeError(
|
|
f"Document Type {document_type.label!r} is referenced and cannot be deleted",
|
|
category=ErrorCategory.CONFLICT,
|
|
suggestion="Deactivate the type instead; historical Documents will retain it.",
|
|
)
|
|
await _session.delete(document_type)
|
|
await self._finalize(session=_session, caller_session=session)
|
|
|
|
async def is_document_type_referenced(
|
|
self,
|
|
document_type_id: UUID,
|
|
*,
|
|
session: AsyncSession | None = None,
|
|
) -> bool:
|
|
"""Return whether a Document references a Document Type."""
|
|
async with self._session_scope(session) as _session:
|
|
document_type = await _session.get(DocumentType, document_type_id)
|
|
if document_type is None:
|
|
raise DocumentTypeError(
|
|
f"Document Type with id {document_type_id} not found",
|
|
category=ErrorCategory.NOT_FOUND,
|
|
suggestion="Refresh Settings and select an available Document Type.",
|
|
)
|
|
return await self._document_type_is_referenced(
|
|
session=_session,
|
|
document_type=document_type,
|
|
)
|
|
|
|
@staticmethod
|
|
async def _document_type_is_referenced(
|
|
*,
|
|
session: AsyncSession,
|
|
document_type: DocumentType,
|
|
) -> bool:
|
|
reference = (
|
|
await session.exec(select(Document.id).where(Document.document_type_id == document_type.id))
|
|
).first()
|
|
return reference is not None
|
|
|
|
async def set_document_type(
|
|
self,
|
|
*,
|
|
document_id: UUID,
|
|
document_type_id: UUID,
|
|
session: AsyncSession | None = None,
|
|
) -> Document:
|
|
"""Set a Document Type by UUID."""
|
|
async with self._session_scope(session) as _session:
|
|
document = await self._get_document_or_raise(session=_session, document_id=document_id)
|
|
document.document_type_id = document_type_id
|
|
await self._validate_document_type(session=_session, document=document)
|
|
document.updated_at = datetime.now(UTC)
|
|
await self._finalize(session=_session, caller_session=session, refresh=(document,))
|
|
return document
|