Coverage for src/paperap/models/document/model.py: 76%
307 statements
« prev ^ index » next coverage.py v7.6.12, created at 2025-03-22 16:02 -0400
1"""
5----------------------------------------------------------------------------
7METADATA:
9File: model.py
10 Project: paperap
11Created: 2025-03-09
12 Version: 0.0.10
13Author: Jess Mann
14Email: jess@jmann.me
15 Copyright (c) 2025 Jess Mann
17----------------------------------------------------------------------------
19LAST MODIFIED:
212025-03-09 By Jess Mann
23"""
25from __future__ import annotations
27import logging
28from datetime import datetime
29from typing import TYPE_CHECKING, Annotated, Any, Iterable, Iterator, TypedDict, cast, override
31import pydantic
32from pydantic import Field, field_serializer, field_validator, model_serializer
33from typing_extensions import TypeVar
35from paperap.const import (
36 CustomFieldTypedDict,
37 CustomFieldTypes,
38 CustomFieldValues,
39 DocumentStorageType,
40 FilteringStrategies,
41)
42from paperap.exceptions import ResourceNotFoundError
43from paperap.models.abstract.model import StandardModel
44from paperap.models.document.meta import SUPPORTED_FILTERING_PARAMS
45from paperap.models.document.queryset import DocumentQuerySet
47if TYPE_CHECKING:
48 from paperap.models.correspondent.model import Correspondent
49 from paperap.models.custom_field import CustomField, CustomFieldQuerySet
50 from paperap.models.document.download.model import DownloadedDocument
51 from paperap.models.document.metadata.model import DocumentMetadata
52 from paperap.models.document.suggestions.model import DocumentSuggestions
53 from paperap.models.document_type.model import DocumentType
54 from paperap.models.storage_path.model import StoragePath
55 from paperap.models.tag import Tag, TagQuerySet
56 from paperap.models.user.model import User
58logger = logging.getLogger(__name__)
61class DocumentNote(StandardModel):
62 """
63 Represents a note on a Paperless-NgX document.
64 """
66 deleted_at: datetime | None = None
67 restored_at: datetime | None = None
68 transaction_id: int | None = None
69 note: str
70 created: datetime
71 document: int
72 user: int
74 class Meta(StandardModel.Meta):
75 read_only_fields = {"deleted_at", "restored_at", "transaction_id", "created"}
77 @field_serializer("deleted_at", "restored_at", "created")
78 def serialize_datetime(self, value: datetime | None) -> str | None:
79 """
80 Serialize datetime fields to ISO format.
82 Args:
83 value: The datetime value to serialize.
85 Returns:
86 The serialized datetime value or None if the value is None.
88 """
89 return value.isoformat() if value else None
91 def get_document(self) -> "Document":
92 """
93 Get the document associated with this note.
95 Returns:
96 The document associated with this note.
98 """
99 return self._client.documents().get(self.document)
101 def get_user(self) -> "User":
102 """
103 Get the user who created this note.
105 Returns:
106 The user who created this note.
108 """
109 return self._client.users().get(self.user)
112class Document(StandardModel):
113 """
114 Represents a Paperless-NgX document.
116 Attributes:
117 added: The timestamp when the document was added to the system.
118 archive_serial_number: The serial number of the archive.
119 archived_file_name: The name of the archived file.
120 content: The content of the document.
121 correspondent: The correspondent associated with the document.
122 created: The timestamp when the document was created.
123 created_date: The date when the document was created.
124 updated: The timestamp when the document was last updated.
125 custom_fields: Custom fields associated with the document.
126 deleted_at: The timestamp when the document was deleted.
127 document_type: The document type associated with the document.
128 is_shared_by_requester: Whether the document is shared by the requester.
129 notes: Notes associated with the document.
130 original_filename: The original file name of the document.
131 owner: The owner of the document.
132 page_count: The number of pages in the document.
133 storage_path: The storage path of the document.
134 tags: The tags associated with the document.
135 title: The title of the document.
136 user_can_change: Whether the user can change the document.
137 checksum: The checksum of the document.
139 Examples:
140 >>> document = client.documents().get(pk=1)
141 >>> document.title = 'Example Document'
142 >>> document.save()
143 >>> document.title
144 'Example Document'
146 # Get document metadata
147 >>> metadata = document.get_metadata()
148 >>> print(metadata.original_mime_type)
150 # Download document
151 >>> download = document.download()
152 >>> with open(download.disposition_filename, 'wb') as f:
153 ... f.write(download.content)
155 # Get document suggestions
156 >>> suggestions = document.get_suggestions()
157 >>> print(suggestions.tags)
159 """
161 # where did this come from? It's not in sample data?
162 added: datetime | None = None
163 archive_checksum: str | None = None
164 archive_filename: str | None = None
165 archive_serial_number: int | None = None
166 archived_file_name: str | None = None
167 checksum: str | None = None
168 content: str = ""
169 correspondent_id: int | None = None
170 created: datetime | None = Field(description="Creation timestamp", default=None)
171 created_date: str | None = None
172 custom_field_dicts: Annotated[list[CustomFieldValues], Field(default_factory=list)]
173 deleted_at: datetime | None = None
174 document_type_id: int | None = None
175 filename: str | None = None
176 is_shared_by_requester: bool = False
177 notes: "list[DocumentNote]" = Field(default_factory=list)
178 original_filename: str | None = None
179 owner: int | None = None
180 page_count: int | None = None
181 storage_path_id: int | None = None
182 storage_type: DocumentStorageType | None = None
183 tag_ids: Annotated[list[int], Field(default_factory=list)]
184 title: str = ""
185 user_can_change: bool | None = None
187 _correspondent: tuple[int, Correspondent] | None = None
188 _document_type: tuple[int, DocumentType] | None = None
189 _storage_path: tuple[int, StoragePath] | None = None
190 __search_hit__: dict[str, Any] | None = None
192 class Meta(StandardModel.Meta):
193 # NOTE: Filtering appears to be disabled by paperless on page_count
194 read_only_fields = {"page_count", "deleted_at", "is_shared_by_requester", "archived_file_name"}
195 filtering_disabled = {"page_count", "deleted_at", "is_shared_by_requester"}
196 filtering_strategies = {FilteringStrategies.WHITELIST}
197 field_map = {
198 "tags": "tag_ids",
199 "custom_fields": "custom_field_dicts",
200 "document_type": "document_type_id",
201 "correspondent": "correspondent_id",
202 "storage_path": "storage_path_id",
203 }
204 supported_filtering_params = SUPPORTED_FILTERING_PARAMS
206 @field_serializer("added", "created", "deleted_at")
207 def serialize_datetime(self, value: datetime | None) -> str | None:
208 """
209 Serialize datetime fields to ISO format.
211 Args:
212 value: The datetime value to serialize.
214 Returns:
215 The serialized datetime value.
217 """
218 return value.isoformat() if value else None
220 @field_serializer("notes")
221 def serialize_notes(self, value: list[DocumentNote]) -> list[dict[str, Any]]:
222 """
223 Serialize notes to a list of dictionaries.
225 Args:
226 value: The list of DocumentNote objects to serialize.
228 Returns:
229 A list of dictionaries representing the notes.
231 """
232 return [note.to_dict() for note in value] if value else []
234 @field_validator("tag_ids", mode="before")
235 @classmethod
236 def validate_tags(cls, value: Any) -> list[int]:
237 """
238 Validate and convert tag IDs to a list of integers.
240 Args:
241 value: The list of tag IDs to validate.
243 Returns:
244 A list of validated tag IDs.
246 """
247 if value is None:
248 return []
250 if isinstance(value, list):
251 return [int(tag) for tag in value]
253 if isinstance(value, int):
254 return [value]
256 raise TypeError(f"Invalid type for tags: {type(value)}")
258 @field_validator("custom_field_dicts", mode="before")
259 @classmethod
260 def validate_custom_fields(cls, value: Any) -> list[CustomFieldValues]:
261 """
262 Validate and return custom field dictionaries.
264 Args:
265 value: The list of custom field dictionaries to validate.
267 Returns:
268 A list of validated custom field dictionaries.
270 """
271 if value is None:
272 return []
274 if isinstance(value, list):
275 return value
277 raise TypeError(f"Invalid type for custom fields: {type(value)}")
279 @field_validator("content", "title", mode="before")
280 @classmethod
281 def validate_text(cls, value: Any) -> str:
282 """
283 Validate and return a text field.
285 Args:
286 value: The value of the text field to validate.
288 Returns:
289 The validated text value.
291 """
292 if value is None:
293 return ""
295 if isinstance(value, (str, int)):
296 return str(value)
298 raise TypeError(f"Invalid type for text: {type(value)}")
300 @field_validator("notes", mode="before")
301 @classmethod
302 def validate_notes(cls, value: Any) -> list[Any]:
303 """
304 Validate and return the list of notes.
306 Args:
307 value: The list of notes to validate.
309 Returns:
310 The validated list of notes.
312 """
313 if value is None:
314 return []
316 if isinstance(value, list):
317 return value
319 if isinstance(value, DocumentNote):
320 return [value]
322 raise TypeError(f"Invalid type for notes: {type(value)}")
324 @field_validator("is_shared_by_requester", mode="before")
325 @classmethod
326 def validate_is_shared_by_requester(cls, value: Any) -> bool:
327 """
328 Validate and return the is_shared_by_requester flag.
330 Args:
331 value: The flag to validate.
333 Returns:
334 The validated flag.
336 """
337 if value is None:
338 return False
340 if isinstance(value, bool):
341 return value
343 raise TypeError(f"Invalid type for is_shared_by_requester: {type(value)}")
345 @property
346 def custom_field_ids(self) -> list[int]:
347 """
348 Get the IDs of the custom fields for this document.
349 """
350 return [element.field for element in self.custom_field_dicts]
352 @property
353 def custom_field_values(self) -> list[Any]:
354 """
355 Get the values of the custom fields for this document.
356 """
357 return [element.value for element in self.custom_field_dicts]
359 @property
360 def tag_names(self) -> list[str]:
361 """
362 Get the names of the tags for this document.
363 """
364 return [tag.name for tag in self.tags if tag.name]
366 @property
367 def tags(self) -> TagQuerySet:
368 """
369 Get the tags for this document.
371 Returns:
372 List of tags associated with this document.
374 Examples:
375 >>> document = client.documents().get(pk=1)
376 >>> for tag in document.tags:
377 ... print(f'{tag.name} # {tag.id}')
378 'Tag 1 # 1'
379 'Tag 2 # 2'
380 'Tag 3 # 3'
382 >>> if 5 in document.tags:
383 ... print('Tag ID #5 is associated with this document')
385 >>> tag = client.tags().get(pk=1)
386 >>> if tag in document.tags:
387 ... print('Tag ID #1 is associated with this document')
389 >>> filtered_tags = document.tags.filter(name__icontains='example')
390 >>> for tag in filtered_tags:
391 ... print(f'{tag.name} # {tag.id}')
393 """
394 if not self.tag_ids:
395 return self._client.tags().none()
397 # Use the API's filtering capability to get only the tags with specific IDs
398 # The paperless-ngx API supports id__in filter for retrieving multiple objects by ID
399 return self._client.tags().id(self.tag_ids)
401 @tags.setter
402 def tags(self, value: "Iterable[Tag | int] | None") -> None:
403 """
404 Set the tags for this document.
406 Args:
407 value: The tags to set.
409 """
410 if value is None:
411 self.tag_ids = []
412 return
414 if isinstance(value, Iterable):
415 # Reset tag_ids to ensure we only have the new values
416 self.tag_ids = []
417 for tag in value:
418 if isinstance(tag, int):
419 self.tag_ids.append(tag)
420 continue
422 # Check against StandardModel to avoid circular imports
423 # If it is another type of standard model, pydantic validators will complain
424 if isinstance(tag, StandardModel):
425 self.tag_ids.append(tag.id)
426 continue
428 raise TypeError(f"Invalid type for tags: {type(tag)}")
429 return
431 raise TypeError(f"Invalid type for tags: {type(value)}")
433 @property
434 def correspondent(self) -> "Correspondent | None":
435 """
436 Get the correspondent for this document.
438 Returns:
439 The correspondent or None if not set.
441 Examples:
442 >>> document = client.documents().get(pk=1)
443 >>> document.correspondent.name
444 'Example Correspondent'
446 """
447 # Return cache
448 if self._correspondent is not None:
449 pk, value = self._correspondent
450 if pk == self.correspondent_id:
451 return value
453 # None set to retrieve
454 if not self.correspondent_id:
455 return None
457 # Retrieve it
458 correspondent = self._client.correspondents().get(self.correspondent_id)
459 self._correspondent = (self.correspondent_id, correspondent)
460 return correspondent
462 @correspondent.setter
463 def correspondent(self, value: "Correspondent | int | None") -> None:
464 """
465 Set the correspondent for this document.
467 Args:
468 value: The correspondent to set.
470 """
471 if value is None:
472 # Leave cache in place in case it changes again
473 self.correspondent_id = None
474 return
476 if isinstance(value, int):
477 # Leave cache in place in case id is the same, or id changes again
478 self.correspondent_id = value
479 return
481 # Check against StandardModel to avoid circular imports
482 # If it is another type of standard model, pydantic validators will complain
483 if isinstance(value, StandardModel):
484 self.correspondent_id = value.id
485 # Pre-populate the cache
486 self._correspondent = (value.id, value)
487 return
489 raise TypeError(f"Invalid type for correspondent: {type(value)}")
491 @property
492 def document_type(self) -> "DocumentType | None":
493 """
494 Get the document type for this document.
496 Returns:
497 The document type or None if not set.
499 Examples:
500 >>> document = client.documents().get(pk=1)
501 >>> document.document_type.name
502 'Example Document Type
504 """
505 # Return cache
506 if self._document_type is not None:
507 pk, value = self._document_type
508 if pk == self.document_type_id:
509 return value
511 # None set to retrieve
512 if not self.document_type_id:
513 return None
515 # Retrieve it
516 document_type = self._client.document_types().get(self.document_type_id)
517 self._document_type = (self.document_type_id, document_type)
518 return document_type
520 @document_type.setter
521 def document_type(self, value: "DocumentType | int | None") -> None:
522 """
523 Set the document type for this document.
525 Args:
526 value: The document type to set.
528 """
529 if value is None:
530 # Leave cache in place in case it changes again
531 self.document_type_id = None
532 return
534 if isinstance(value, int):
535 # Leave cache in place in case id is the same, or id changes again
536 self.document_type_id = value
537 return
539 # Check against StandardModel to avoid circular imports
540 # If it is another type of standard model, pydantic validators will complain
541 if isinstance(value, StandardModel):
542 self.document_type_id = value.id
543 # Pre-populate the cache
544 self._document_type = (value.id, value)
545 return
547 raise TypeError(f"Invalid type for document_type: {type(value)}")
549 @property
550 def storage_path(self) -> "StoragePath | None":
551 """
552 Get the storage path for this document.
554 Returns:
555 The storage path or None if not set.
557 Examples:
558 >>> document = client.documents().get(pk=1)
559 >>> document.storage_path.name
560 'Example Storage Path'
562 """
563 # Return cache
564 if self._storage_path is not None:
565 pk, value = self._storage_path
566 if pk == self.storage_path_id:
567 return value
569 # None set to retrieve
570 if not self.storage_path_id:
571 return None
573 # Retrieve it
574 storage_path = self._client.storage_paths().get(self.storage_path_id)
575 self._storage_path = (self.storage_path_id, storage_path)
576 return storage_path
578 @storage_path.setter
579 def storage_path(self, value: "StoragePath | int | None") -> None:
580 """
581 Set the storage path for this document.
583 Args:
584 value: The storage path to set.
586 """
587 if value is None:
588 # Leave cache in place in case it changes again
589 self.storage_path_id = None
590 return
592 if isinstance(value, int):
593 # Leave cache in place in case id is the same, or id changes again
594 self.storage_path_id = value
595 return
597 # Check against StandardModel to avoid circular imports
598 # If it is another type of standard model, pydantic validators will complain
599 if isinstance(value, StandardModel):
600 self.storage_path_id = value.id
601 # Pre-populate the cache
602 self._storage_path = (value.id, value)
603 return
605 raise TypeError(f"Invalid type for storage_path: {type(value)}")
607 @property
608 def custom_fields(self) -> "CustomFieldQuerySet":
609 """
610 Get the custom fields for this document.
612 Returns:
613 List of custom fields associated with this document.
615 """
616 if not self.custom_field_dicts:
617 return self._client.custom_fields().none()
619 # Use the API's filtering capability to get only the custom fields with specific IDs
620 # The paperless-ngx API supports id__in filter for retrieving multiple objects by ID
621 return self._client.custom_fields().id(self.custom_field_ids)
623 @custom_fields.setter
624 def custom_fields(self, value: "Iterable[CustomField | CustomFieldValues | CustomFieldTypedDict] | None") -> None:
625 """
626 Set the custom fields for this document.
628 Args:
629 value: The custom fields to set.
631 """
632 if value is None:
633 self.custom_field_dicts = []
634 return
636 if isinstance(value, Iterable):
637 new_list: list[CustomFieldValues] = []
638 for field in value:
639 if isinstance(field, CustomFieldValues):
640 new_list.append(field)
641 continue
643 # isinstance(field, CustomField)
644 # Check against StandardModel (instead of CustomField) to avoid circular imports
645 # If it is the wrong type of standard model (e.g. a User), pydantic validators will complain
646 if isinstance(field, StandardModel):
647 new_list.append(CustomFieldValues(field=field.id, value=getattr(field, "value")))
648 continue
650 if isinstance(field, dict):
651 new_list.append(CustomFieldValues(**field))
652 continue
654 raise TypeError(f"Invalid type for custom fields: {type(field)}")
656 self.custom_field_dicts = new_list
657 return
659 raise TypeError(f"Invalid type for custom fields: {type(value)}")
661 @property
662 def has_search_hit(self) -> bool:
663 return self.__search_hit__ is not None
665 @property
666 def search_hit(self) -> dict[str, Any] | None:
667 return self.__search_hit__
669 def custom_field_value(self, field_id: int, default: Any = None, *, raise_errors: bool = False) -> Any:
670 """
671 Get the value of a custom field by ID.
673 Args:
674 field_id: The ID of the custom field.
675 default: The value to return if the field is not found.
676 raise_errors: Whether to raise an error if the field is not found.
678 Returns:
679 The value of the custom field or the default value if not found.
681 """
682 for field in self.custom_field_dicts:
683 if field.field == field_id:
684 return field.value
686 if raise_errors:
687 raise ValueError(f"Custom field {field_id} not found")
688 return default
690 """
691 def __getattr__(self, name: str) -> Any:
692 # Allow easy access to custom fields
693 for custom_field in self.custom_fields:
694 if custom_field['field'] == name:
695 return custom_field['value']
697 raise AttributeError(f"'{self.__class__.__name__}' object has no attribute '{name}'")
698 """
700 def add_tag(self, tag: "Tag | int | str") -> None:
701 """
702 Add a tag to the document.
704 Args:
705 tag: The tag to add.
707 """
708 if isinstance(tag, int):
709 self.tag_ids.append(tag)
710 return
712 if isinstance(tag, StandardModel):
713 self.tag_ids.append(tag.id)
714 return
716 if isinstance(tag, str):
717 if not (instance := self._client.tags().filter(name=tag).first()):
718 raise ResourceNotFoundError(f"Tag '{tag}' not found")
719 self.tag_ids.append(instance.id)
720 return
722 raise TypeError(f"Invalid type for tag: {type(tag)}")
724 def remove_tag(self, tag: "Tag | int | str") -> None:
725 """
726 Remove a tag from the document.
728 Args:
729 tag: The tag to remove.
731 """
732 if isinstance(tag, int):
733 # TODO: Handle removal with consideration of "tags can't be empty" rule in paperless
734 self.tag_ids.remove(tag)
735 return
737 if isinstance(tag, StandardModel):
738 # TODO: Handle removal with consideration of "tags can't be empty" rule in paperless
739 self.tag_ids.remove(tag.id)
740 return
742 if isinstance(tag, str):
743 # TODO: Handle removal with consideration of "tags can't be empty" rule in paperless
744 if not (instance := self._client.tags().filter(name=tag).first()):
745 raise ResourceNotFoundError(f"Tag '{tag}' not found")
746 self.tag_ids.remove(instance.id)
747 return
749 raise TypeError(f"Invalid type for tag: {type(tag)}")
751 def get_metadata(self) -> "DocumentMetadata":
752 """
753 Get the metadata for this document.
755 Returns:
756 The document metadata.
758 Examples:
759 >>> metadata = document.get_metadata()
760 >>> print(metadata.original_mime_type)
762 """
763 raise NotImplementedError()
765 def download(self, original: bool = False) -> "DownloadedDocument":
766 """
767 Download the document file.
769 Args:
770 original: Whether to download the original file instead of the archived version.
772 Returns:
773 The downloaded document.
775 Examples:
776 >>> download = document.download()
777 >>> with open(download.disposition_filename, 'wb') as f:
778 ... f.write(download.content)
780 """
781 raise NotImplementedError()
783 def preview(self, original: bool = False) -> "DownloadedDocument":
784 """
785 Get a preview of the document.
787 Args:
788 original: Whether to preview the original file instead of the archived version.
790 Returns:
791 The document preview.
793 """
794 raise NotImplementedError()
796 def thumbnail(self, original: bool = False) -> "DownloadedDocument":
797 """
798 Get the document thumbnail.
800 Args:
801 original: Whether to get the thumbnail of the original file.
803 Returns:
804 The document thumbnail.
806 """
807 raise NotImplementedError()
809 def get_suggestions(self) -> "DocumentSuggestions":
810 """
811 Get suggestions for this document.
813 Returns:
814 The document suggestions.
816 Examples:
817 >>> suggestions = document.get_suggestions()
818 >>> print(suggestions.tags)
820 """
821 raise NotImplementedError()
823 def append_content(self, value: str) -> None:
824 """
825 Append content to the document.
827 Args:
828 value: The content to append.
830 """
831 self.content = f"{self.content}\n{value}"
833 @override
834 def update_locally(self, from_db: bool | None = None, **kwargs: Any) -> None:
835 """
836 Update the document locally with the provided data.
838 Args:
839 from_db: Whether to update from the database.
840 **kwargs: Additional data to update the document with.
842 Raises:
843 NotImplementedError: If attempting to set notes or tags to None when they are not already None.
845 """
846 if not from_db:
847 # Paperless does not support setting notes or tags to None if not already None
848 fields = ["notes", "tag_ids"]
849 for field in fields:
850 original = self._original_data[field]
851 if original and field in kwargs and not kwargs.get(field):
852 raise NotImplementedError(f"Cannot set {field} to None. {field} currently: {original}")
854 # Handle aliases
855 if self._original_data["tag_ids"] and "tags" in kwargs and not kwargs.get("tags"):
856 raise NotImplementedError(f"Cannot set tags to None. Tags currently: {self._original_data['tag_ids']}")
858 return super().update_locally(from_db=from_db, **kwargs)