mirror of
https://github.com/paperless-ngx/paperless-ngx.git
synced 2026-04-18 14:08:51 +00:00
- Replace Document-as-version pattern with DocumentVersion objects - Add _create_doc and _create_version helpers; remove _create_pdf - Remove deleted root/ endpoint tests - Fix views.py: use isinstance(content_doc, DocumentVersion) instead of id comparison (cross-table id collision), add Document.content sync when latest version content is edited - Add compatibility stubs to versioning.py (get_root_document, get_latest_version_for_root, resolve_effective_document_by_pk, EffectiveDocumentResolution) so bulk_edit.py and conditionals.py imports resolve; Tasks 9 and 10 will refactor the callers - Add DocumentVersion.modified property shim for conditionals.py Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com>
120 lines
3.7 KiB
Python
120 lines
3.7 KiB
Python
from __future__ import annotations
|
|
|
|
from dataclasses import dataclass
|
|
from enum import StrEnum
|
|
from typing import TYPE_CHECKING
|
|
from typing import Any
|
|
|
|
from documents.models import Document
|
|
from documents.models import DocumentVersion
|
|
|
|
if TYPE_CHECKING:
|
|
from django.http import HttpRequest
|
|
|
|
|
|
class VersionResolutionError(StrEnum):
|
|
INVALID = "invalid"
|
|
NOT_FOUND = "not_found"
|
|
|
|
|
|
@dataclass(frozen=True, slots=True)
|
|
class VersionResolution:
|
|
version: DocumentVersion | None
|
|
error: VersionResolutionError | None = None
|
|
|
|
|
|
def get_request_version_param(request: HttpRequest) -> str | None:
|
|
if hasattr(request, "query_params"):
|
|
return request.query_params.get("version")
|
|
return None
|
|
|
|
|
|
def get_latest_version(doc: Document) -> DocumentVersion | None:
|
|
"""Return the highest-version_number DocumentVersion for doc, or None."""
|
|
return (
|
|
DocumentVersion.objects.filter(document=doc).order_by("-version_number").first()
|
|
)
|
|
|
|
|
|
def get_version_by_pk(doc: Document, version_pk: int) -> DocumentVersion | None:
|
|
"""Return the DocumentVersion with the given pk if it belongs to doc."""
|
|
return DocumentVersion.objects.filter(pk=version_pk, document=doc).first()
|
|
|
|
|
|
@dataclass(frozen=True, slots=True)
|
|
class EffectiveDocumentResolution:
|
|
"""Compatibility shim for conditionals.py callers that access .document.
|
|
|
|
Task 9 will refactor these callers to use VersionResolution.version directly.
|
|
"""
|
|
|
|
document: DocumentVersion | None
|
|
|
|
|
|
def resolve_effective_document_by_pk(
|
|
pk: int,
|
|
request: Any,
|
|
) -> EffectiveDocumentResolution:
|
|
"""Resolve the effective DocumentVersion by document pk and request params.
|
|
|
|
This is a compatibility stub used by conditionals.py; Task 9 will refactor
|
|
the callers to use resolve_requested_version directly.
|
|
"""
|
|
try:
|
|
doc = Document.objects.get(pk=pk)
|
|
except Document.DoesNotExist:
|
|
return EffectiveDocumentResolution(document=None)
|
|
resolution = resolve_requested_version(doc, request)
|
|
return EffectiveDocumentResolution(document=resolution.version)
|
|
|
|
|
|
def get_root_document(doc: Document) -> Document:
|
|
"""Return the root document.
|
|
|
|
In the new model, every Document IS the root. This is a compatibility stub
|
|
used by bulk_edit.py; Task 10 will refactor the callers.
|
|
"""
|
|
return doc
|
|
|
|
|
|
def get_latest_version_for_root(doc: Document) -> Document:
|
|
"""Return the document to use as the version source.
|
|
|
|
In the new model, DocumentVersion holds per-version files. This stub
|
|
returns the document itself so that bulk_edit callers that have not yet
|
|
been updated to the new model do not crash. Task 10 will replace this.
|
|
"""
|
|
return doc
|
|
|
|
|
|
def resolve_requested_version(
|
|
doc: Document,
|
|
request: Any,
|
|
) -> VersionResolution:
|
|
"""
|
|
Resolve the DocumentVersion to serve based on the optional ``?version=<pk>``
|
|
query parameter.
|
|
|
|
- No parameter: return the latest version.
|
|
- Parameter present: validate and return that specific version.
|
|
"""
|
|
version_param = get_request_version_param(request)
|
|
if not version_param:
|
|
latest = get_latest_version(doc)
|
|
if latest is None:
|
|
return VersionResolution(
|
|
version=None,
|
|
error=VersionResolutionError.NOT_FOUND,
|
|
)
|
|
return VersionResolution(version=latest)
|
|
|
|
try:
|
|
version_pk = int(version_param)
|
|
except (TypeError, ValueError):
|
|
return VersionResolution(version=None, error=VersionResolutionError.INVALID)
|
|
|
|
version = get_version_by_pk(doc, version_pk)
|
|
if version is None:
|
|
return VersionResolution(version=None, error=VersionResolutionError.NOT_FOUND)
|
|
return VersionResolution(version=version)
|