Coverage for documents/versioning.py: 49%
107 statements
« prev ^ index » next coverage.py v7.15.2, created at 2026-10-10 09:07 +0000
1from __future__ import annotations
3from dataclasses import dataclass
4from enum import StrEnum
5from typing import TYPE_CHECKING
6from typing import Any
8from django.db.models import F
9from django.db.models import OuterRef
10from django.db.models import Prefetch
11from django.db.models import QuerySet
12from django.db.models import Subquery
13from django.db.models import Window
14from django.db.models.functions import Coalesce
15from django.db.models.functions import RowNumber
17from documents.models import Document
19if TYPE_CHECKING: 19 ↛ 20line 19 didn't jump to line 20 because the condition on line 19 was never true
20 from rest_framework.request import Request
23def versions_newest_first(documents: QuerySet[Document]) -> QuerySet[Document]:
24 """
25 Sorts versions so the newest one comes first using version_index and not on id,
26 because an existing document can be merged in as a version
27 """
28 return documents.order_by(F("version_index").desc(nulls_last=True), "-id")
31def annotate_effective_content(documents: QuerySet[Document]) -> QuerySet[Document]:
32 """
33 Annotates documents with the content of their newest version unless the
34 queryset already carries the annotation, falling back to their own, so
35 get_effective_content() can answer from the row rather than querying for
36 the versions of each document.
37 """
38 if "effective_content" in documents.query.annotations:
39 return documents
40 return documents.annotate(
41 effective_content=Coalesce(
42 Subquery(
43 versions_newest_first(
44 Document.objects.filter(root_document=OuterRef("pk")),
45 ).values("content")[:1],
46 ),
47 F("content"),
48 ),
49 )
52LATEST_VERSION_CONTENT_PREFETCH_ATTR = "_latest_version_content_prefetch"
55def latest_version_content_prefetch() -> Prefetch:
56 """
57 A Prefetch for Document.versions scoped to just the newest version's
58 content, for get_effective_content()'s fallback when no SQL annotation
59 is present.
61 Deliberately not merged into a metadata-only "versions" prefetch (the one
62 used for the serialized versions list): that one fetches every historical
63 version of every document, and pulling full OCR content for versions
64 nobody will read wastes DB transfer/memory at scale. This one is windowed
65 down to a single row per root, then bounded by Prefetch's own IN-list to
66 whatever page/result set it's attached to -- one cheap bulk query total,
67 not one per document and not one per version.
68 """
69 return Prefetch(
70 "versions",
71 queryset=(
72 Document.objects.filter(
73 root_document_id__isnull=False,
74 deleted_at__isnull=True,
75 )
76 .annotate(
77 rn=Window(
78 RowNumber(),
79 partition_by=F("root_document_id"),
80 order_by=[
81 F("version_index").desc(nulls_last=True),
82 F("id").desc(),
83 ],
84 ),
85 )
86 .filter(rn=1)
87 .only("id", "root_document_id", "content")
88 ),
89 to_attr=LATEST_VERSION_CONTENT_PREFETCH_ATTR,
90 )
93def has_prefetched_effective_content(document: Document) -> bool:
94 """
95 True if document.get_effective_content() can answer without an extra
96 per-instance query -- an SQL ``effective_content`` annotation, the lean
97 latest_version_content_prefetch(), or the metadata-only "versions"
98 prefetch is already present on the instance.
100 Callers that haven't set any of those up (e.g. views that build their
101 own querysets independently of DocumentViewSet.get_queryset(), like
102 TrashView or GlobalSearchView) intentionally don't pay for version-aware
103 content resolution -- see DocumentSerializer.to_representation(), which
104 uses this to decide whether to call get_effective_content() at all.
105 """
106 if hasattr(document, "effective_content"):
107 return True
108 if getattr(document, LATEST_VERSION_CONTENT_PREFETCH_ATTR, None) is not None:
109 return True
110 prefetched_cache = getattr(document, "_prefetched_objects_cache", None)
111 return isinstance(prefetched_cache, dict) and "versions" in prefetched_cache
114def sort_versions_newest_first(documents: list[Document]) -> list[Document]:
115 """
116 Same sorting as versions_newest_first()
117 """
118 return sorted(
119 documents,
120 key=lambda doc: (doc.version_index or 0, doc.id),
121 reverse=True,
122 )
125class VersionResolutionError(StrEnum):
126 INVALID = "invalid"
127 NOT_FOUND = "not_found"
130@dataclass(frozen=True, slots=True)
131class VersionResolution:
132 document: Document | None
133 error: VersionResolutionError | None = None
136def _document_manager(*, include_deleted: bool) -> Any:
137 return Document.global_objects if include_deleted else Document.objects
140def get_request_version_param(request: Request) -> str | None:
141 if hasattr(request, "query_params"):
142 return request.query_params.get("version")
143 return None
146def get_root_document(doc: Document, *, include_deleted: bool = False) -> Document:
147 # Use root_document_id to avoid a query when this is already a root.
148 # If root_document isn't available, fall back to the document itself.
149 if doc.root_document_id is None: 149 ↛ 151line 149 didn't jump to line 151 because the condition on line 149 was always true
150 return doc
151 if doc.root_document is not None:
152 return doc.root_document
154 manager = _document_manager(include_deleted=include_deleted)
155 root_doc = manager.only("id").filter(id=doc.root_document_id).first()
156 return root_doc or doc
159def get_latest_version_for_root(
160 root_doc: Document,
161 *,
162 include_deleted: bool = False,
163) -> Document:
164 manager = _document_manager(include_deleted=include_deleted)
165 latest = versions_newest_first(manager.filter(root_document=root_doc)).first()
166 return latest or root_doc
169def latest_version(document: Document) -> Document:
170 """
171 The newest version of a root document, or the document itself if it has
172 no versions or is a version. Reads a prefetched "versions" lookup when
173 there is one and queries otherwise.
174 """
175 if document.root_document_id is not None or document.pk is None:
176 return document
177 prefetched_cache = getattr(document, "_prefetched_objects_cache", None)
178 prefetched_versions = (
179 prefetched_cache.get("versions") if isinstance(prefetched_cache, dict) else None
180 )
181 if prefetched_versions is None:
182 return get_latest_version_for_root(document)
183 if not prefetched_versions:
184 return document
185 return sort_versions_newest_first(list(prefetched_versions))[0]
188def resolve_requested_version_for_root(
189 root_doc: Document,
190 request: Request,
191 *,
192 include_deleted: bool = False,
193) -> VersionResolution:
194 version_param = get_request_version_param(request)
195 if not version_param:
196 return VersionResolution(
197 document=get_latest_version_for_root(
198 root_doc,
199 include_deleted=include_deleted,
200 ),
201 )
203 try:
204 version_id = int(version_param)
205 except (TypeError, ValueError):
206 return VersionResolution(document=None, error=VersionResolutionError.INVALID)
208 manager = _document_manager(include_deleted=include_deleted)
209 candidate = manager.only("id", "root_document_id").filter(id=version_id).first()
210 if candidate is None:
211 return VersionResolution(document=None, error=VersionResolutionError.NOT_FOUND)
212 if candidate.id != root_doc.id and candidate.root_document_id != root_doc.id:
213 return VersionResolution(document=None, error=VersionResolutionError.NOT_FOUND)
214 return VersionResolution(document=candidate)
217def resolve_effective_document(
218 request_doc: Document,
219 request: Request,
220 *,
221 include_deleted: bool = False,
222) -> VersionResolution:
223 root_doc = get_root_document(request_doc, include_deleted=include_deleted)
224 if get_request_version_param(request) is not None:
225 return resolve_requested_version_for_root(
226 root_doc,
227 request,
228 include_deleted=include_deleted,
229 )
230 if request_doc.root_document_id is None:
231 return VersionResolution(
232 document=get_latest_version_for_root(
233 root_doc,
234 include_deleted=include_deleted,
235 ),
236 )
237 return VersionResolution(document=request_doc)
240_EFFECTIVE_DOCUMENT_CACHE_ATTR = "_effective_document_resolution_cache"
243def resolve_effective_document_by_pk(
244 pk: int,
245 request: Request,
246 *,
247 include_deleted: bool = False,
248) -> VersionResolution:
249 # Django's `condition()` decorator (used for ETag/Last-Modified) invokes the
250 # etag_func and last_modified_func separately, and the view itself may resolve
251 # again -- all against the same request. Cache per-request so a single thumb/
252 # metadata/preview request doesn't redo this resolution multiple times.
253 cache = getattr(request, _EFFECTIVE_DOCUMENT_CACHE_ATTR, None)
254 if cache is None:
255 cache = {}
256 setattr(request, _EFFECTIVE_DOCUMENT_CACHE_ATTR, cache)
258 key = (pk, include_deleted)
259 if key in cache:
260 return cache[key]
262 manager = _document_manager(include_deleted=include_deleted)
263 request_doc = manager.only("id", "root_document_id").filter(pk=pk).first()
264 if request_doc is None: 264 ↛ 270line 264 didn't jump to line 270 because the condition on line 264 was always true
265 resolution = VersionResolution(
266 document=None,
267 error=VersionResolutionError.NOT_FOUND,
268 )
269 else:
270 resolution = resolve_effective_document(
271 request_doc,
272 request,
273 include_deleted=include_deleted,
274 )
276 cache[key] = resolution
277 return resolution