Coverage for src/paperap/resources/documents.py: 55%
141 statements
« prev ^ index » next coverage.py v7.6.12, created at 2025-03-22 16:02 -0400
1"""
2----------------------------------------------------------------------------
4 METADATA:
6 File: documents.py
7 Project: paperap
8 Created: 2025-03-04
9 Version: 0.0.9
10 Author: Jess Mann
11 Email: jess@jmann.me
12 Copyright (c) 2025 Jess Mann
14----------------------------------------------------------------------------
16 LAST MODIFIED:
18 2025-03-04 By Jess Mann
20"""
22from __future__ import annotations
24import logging
25import time
26from datetime import datetime
27from pathlib import Path
28from string import Template
29from typing import Any, Iterator, override
31from typing_extensions import TypeVar
33from paperap.const import URLS
34from paperap.exceptions import APIError, BadResponseError, ResourceNotFoundError
35from paperap.models.document import Document, DocumentNote, DocumentNoteQuerySet, DocumentQuerySet
36from paperap.models.task import Task
37from paperap.resources.base import BaseResource, StandardResource
38from paperap.signals import registry
40logger = logging.getLogger(__name__)
43class DocumentResource(StandardResource[Document, DocumentQuerySet]):
44 """Resource for managing documents."""
46 model_class = Document
47 queryset_class = DocumentQuerySet
48 name = "documents"
49 endpoints = {
50 "list": URLS.list,
51 "detail": URLS.detail,
52 "create": URLS.create,
53 "update": URLS.update,
54 "delete": URLS.delete,
55 "download": URLS.download,
56 "preview": URLS.preview,
57 "thumbnail": URLS.thumbnail,
58 # The upload endpoint does not follow the standard pattern, so we define it explicitly.
59 "upload": Template("/api/documents/post_document/"),
60 "next_asn": URLS.next_asn,
61 "empty_trash": Template("/api/trash/empty/"),
62 }
64 def download(self, document_id: int, *, original: bool = False) -> bytes:
65 url = self.get_endpoint("download", pk=document_id)
66 params = {"original": str(original).lower()}
67 # Request raw bytes by setting json_response to False
68 response = self.client.request("GET", url, params=params, json_response=False)
69 if not response:
70 raise ResourceNotFoundError(f"Document {document_id} download failed", self.name)
71 return response
73 def preview(self, document_id: int) -> bytes:
74 url = self.get_endpoint("preview", pk=document_id)
75 response = self.client.request("GET", url, json_response=False)
76 if response is None:
77 raise ResourceNotFoundError(f"Document {document_id} preview failed", self.name)
78 return response
80 def thumbnail(self, document_id: int) -> bytes:
81 url = self.get_endpoint("thumbnail", pk=document_id)
82 response = self.client.request("GET", url, json_response=False)
83 if response is None:
84 raise ResourceNotFoundError(f"Document {document_id} thumbnail failed", self.name)
85 return response
87 def upload_async(self, filepath: Path | str, **metadata) -> str:
88 """
89 Upload a document from a file to paperless ngx.
91 Args:
92 filepath: The path to the file to upload.
94 Returns:
95 A UUID string (task identifier) as returned by Paperless ngx.
96 e.g. ca6a6dc8-b434-4fcd-8436-8b2546465622
98 Raises:
99 FileNotFoundError: If the file does not exist.
100 ResourceNotFoundError: If the upload fails.
102 """
103 if not isinstance(filepath, Path):
104 filepath = Path(filepath)
105 with filepath.open("rb") as f:
106 return self.upload_content(f.read(), filepath.name, **metadata)
108 def upload_sync(self, filepath: Path | str, max_wait: int = 300, poll_interval: float = 1.0, **metadata) -> Document:
109 """
110 Upload a document and wait until it has been processed.
112 Args:
113 filepath: Path to the file to upload.
114 max_wait: Maximum time (in seconds) to wait for processing.
115 poll_interval: Seconds between polling attempts.
116 **metadata: Additional metadata for the upload.
118 Returns:
119 A Document instance once available.
121 Raises:
122 APIError: If the document is not processed within the max_wait.
123 BadResponseError: If document processing succeeds but no document ID is returned.
125 """
126 task_id = self.upload_async(filepath, **metadata)
127 logger.debug("Upload async complete, task id: %s", task_id)
129 # Define a success callback to handle document retrieval
130 def on_success(task: Task) -> None:
131 if not task.related_document:
132 raise BadResponseError("Document processing succeeded but no document ID was returned")
134 # Wait for the task to complete
135 task = self.client.tasks.wait_for_task(task_id, max_wait=max_wait, poll_interval=poll_interval, success_callback=on_success)
137 if not task.related_document:
138 raise BadResponseError("Document processing succeeded but no document ID was returned")
140 return self.get(task.related_document)
142 def upload_content(self, file_content: bytes, filename: str, **metadata) -> str:
143 """
144 Upload a document with optional metadata.
146 Args:
147 file_content: The binary content of the file to upload
148 filename: The name of the file
149 **metadata: Additional metadata to include with the upload
151 Returns:
152 A string that looks like this: ca6a6dc8-b434-4fcd-8436-8b2546465622
153 This is likely a task id, or similar.
155 Raises:
156 ResourceNotFoundError: If the upload fails
158 """
159 files = {"document": (filename, file_content)}
160 endpoint = self.get_endpoint("upload")
161 response = self.client.request("POST", endpoint, files=files, data=metadata, json_response=True)
162 if not response:
163 raise ResourceNotFoundError("Document upload failed", self.name)
164 return str(response)
166 def next_asn(self) -> int:
167 url = self.get_endpoint("next_asn")
168 response = self.client.request("GET", url)
169 if not response or "next_asn" not in response:
170 raise APIError("Failed to retrieve next ASN")
171 return response["next_asn"]
173 def bulk_action(self, action: str, ids: list[int], **kwargs: Any) -> dict[str, Any]:
174 """
175 Perform a bulk action on multiple documents.
177 Args:
178 action: The action to perform (e.g., "delete", "set_correspondent", etc.)
179 ids: List of document IDs to perform the action on
180 **kwargs: Additional parameters for the action
182 Returns:
183 The API response
185 Raises:
186 ConfigurationError: If the bulk edit endpoint is not defined
188 """
189 # Signal before bulk action
190 signal_params = {"resource": self.name, "action": action, "ids": ids, "kwargs": kwargs}
191 registry.emit("resource.bulk_action:before", "Emitted before bulk action", args=[self], kwargs=signal_params)
193 # Prepare the data for the bulk action
194 data = {"method": action, "documents": ids, "parameters": kwargs}
196 bulk_edit_url = f"{self.client.base_url}/api/documents/bulk_edit/"
197 response = self.client.request("POST", bulk_edit_url, data=data)
199 # Signal after bulk action
200 registry.emit(
201 "resource.bulk_action:after",
202 "Emitted after bulk action",
203 args=[self],
204 kwargs={**signal_params, "response": response},
205 )
207 return response or {}
209 def bulk_delete(self, ids: list[int]) -> dict[str, Any]:
210 """
211 Delete multiple documents at once.
213 Args:
214 ids: List of document IDs to delete
216 Returns:
217 The API response
219 """
220 return self.bulk_action("delete", ids)
222 def bulk_reprocess(self, ids: list[int]) -> dict[str, Any]:
223 """
224 Reprocess multiple documents.
226 Args:
227 ids: List of document IDs to reprocess
229 Returns:
230 The API response
232 """
233 return self.bulk_action("reprocess", ids)
235 def bulk_merge(self, ids: list[int], metadata_document_id: int | None = None, delete_originals: bool = False) -> dict[str, Any]:
236 """
237 Merge multiple documents.
239 Args:
240 ids: List of document IDs to merge
241 metadata_document_id: Apply metadata from this document to the merged document
242 delete_originals: Whether to delete the original documents after merging
244 Returns:
245 The API response
247 """
248 params = {}
249 if metadata_document_id is not None:
250 params["metadata_document_id"] = metadata_document_id
251 if delete_originals:
252 params["delete_originals"] = True
254 return self.bulk_action("merge", ids, **params)
256 def bulk_split(self, document_id: int, pages: list, delete_originals: bool = False) -> dict[str, Any]:
257 """
258 Split a document.
260 Args:
261 document_id: Document ID to split
262 pages: List of pages to split (can include ranges, e.g. "[1,2-3,4,5-7]")
263 delete_originals: Whether to delete the original document after splitting
265 Returns:
266 The API response
268 """
269 params: dict[str, Any] = {"pages": pages}
270 if delete_originals:
271 params["delete_originals"] = True
273 return self.bulk_action("split", [document_id], **params)
275 def bulk_rotate(self, ids: list[int], degrees: int) -> dict[str, Any]:
276 """
277 Rotate documents.
279 Args:
280 ids: List of document IDs to rotate
281 degrees: Degrees to rotate (must be 90, 180, or 270)
283 Returns:
284 The API response
286 """
287 if degrees not in (90, 180, 270):
288 raise ValueError("Degrees must be 90, 180, or 270")
290 return self.bulk_action("rotate", ids, degrees=degrees)
292 def bulk_delete_pages(self, document_id: int, pages: list[int]) -> dict[str, Any]:
293 """
294 Delete pages from a document.
296 Args:
297 document_id: Document ID
298 pages: List of page numbers to delete
300 Returns:
301 The API response
303 """
304 return self.bulk_action("delete_pages", [document_id], pages=pages)
306 def bulk_modify_custom_fields(
307 self,
308 ids: list[int],
309 add_custom_fields: dict[int, Any] | None = None,
310 remove_custom_fields: list[int] | None = None,
311 ) -> dict[str, Any]:
312 """
313 Modify custom fields on multiple documents.
315 Args:
316 ids: List of document IDs to update
317 add_custom_fields: Dictionary of custom field ID to value pairs to add
318 remove_custom_fields: List of custom field IDs to remove
320 Returns:
321 The API response
323 """
324 params: dict[str, Any] = {}
325 if add_custom_fields:
326 params["add_custom_fields"] = add_custom_fields
327 if remove_custom_fields:
328 params["remove_custom_fields"] = remove_custom_fields
330 return self.bulk_action("modify_custom_fields", ids, **params)
332 def bulk_modify_tags(self, ids: list[int], add_tags: list[int] | None = None, remove_tags: list[int] | None = None) -> dict[str, Any]:
333 """
334 Modify tags on multiple documents.
336 Args:
337 ids: List of document IDs to update
338 add_tags: List of tag IDs to add
339 remove_tags: List of tag IDs to remove
341 Returns:
342 The API response
344 """
345 params = {}
346 if add_tags:
347 params["add_tags"] = add_tags
348 if remove_tags:
349 params["remove_tags"] = remove_tags
351 return self.bulk_action("modify_tags", ids, **params)
353 def bulk_add_tag(self, ids: list[int], tag_id: int) -> dict[str, Any]:
354 """
355 Add a tag to multiple documents.
357 Args:
358 ids: List of document IDs to update
359 tag_id: Tag ID to add
361 Returns:
362 The API response
364 """
365 return self.bulk_action("add_tag", ids, tag=tag_id)
367 def bulk_remove_tag(self, ids: list[int], tag_id: int) -> dict[str, Any]:
368 """
369 Remove a tag from multiple documents.
371 Args:
372 ids: List of document IDs to update
373 tag_id: Tag ID to remove
375 Returns:
376 The API response
378 """
379 return self.bulk_action("remove_tag", ids, tag=tag_id)
381 def bulk_set_correspondent(self, ids: list[int], correspondent_id: int) -> dict[str, Any]:
382 """
383 Set correspondent for multiple documents.
385 Args:
386 ids: List of document IDs to update
387 correspondent_id: Correspondent ID to assign
389 Returns:
390 The API response
392 """
393 return self.bulk_action("set_correspondent", ids, correspondent=correspondent_id)
395 def bulk_set_document_type(self, ids: list[int], document_type_id: int) -> dict[str, Any]:
396 """
397 Set document type for multiple documents.
399 Args:
400 ids: List of document IDs to update
401 document_type_id: Document type ID to assign
403 Returns:
404 The API response
406 """
407 return self.bulk_action("set_document_type", ids, document_type=document_type_id)
409 def bulk_set_storage_path(self, ids: list[int], storage_path_id: int) -> dict[str, Any]:
410 """
411 Set storage path for multiple documents.
413 Args:
414 ids: List of document IDs to update
415 storage_path_id: Storage path ID to assign
417 Returns:
418 The API response
420 """
421 return self.bulk_action("set_storage_path", ids, storage_path=storage_path_id)
423 def bulk_set_permissions(
424 self, ids: list[int], permissions: dict[str, Any] | None = None, owner_id: int | None = None, merge: bool = False
425 ) -> dict[str, Any]:
426 """
427 Set permissions for multiple documents.
429 Args:
430 ids: List of document IDs to update
431 permissions: Permissions object
432 owner_id: Owner ID to assign
433 merge: Whether to merge with existing permissions (True) or replace them (False)
435 Returns:
436 The API response
438 """
439 params: dict[str, Any] = {"merge": merge}
440 if permissions:
441 params["set_permissions"] = permissions
442 if owner_id is not None:
443 params["owner"] = owner_id
445 return self.bulk_action("set_permissions", ids, **params)
447 def empty_trash(self) -> dict[str, Any]:
448 """
449 Empty the trash.
451 Returns:
452 The API response.
454 Raises:
455 APIError: If the empty trash request fails.
457 """
458 endpoint = self.get_endpoint("empty_trash")
459 logger.debug("Emptying trash")
460 payload = {"action": "empty"}
461 response = self.client.request("POST", endpoint, data=payload, json_response=True)
462 if not response:
463 raise APIError("Empty trash failed")
464 return response # type: ignore # request should have returned correct response TODO
467class DocumentNoteResource(StandardResource[DocumentNote, DocumentNoteQuerySet]):
468 """Resource for managing document notes."""
470 model_class = DocumentNote
471 queryset_class = DocumentNoteQuerySet
472 name = "notes"
473 endpoints = {"list": Template("/api/document/${pk}/notes/")}