Coverage for documents/serialisers.py: 63%
1768 statements
« prev ^ index » next coverage.py v7.15.2, created at 2026-10-10 09:07 +0000
1from __future__ import annotations
3import logging
4import math
5import re
6from datetime import datetime
7from datetime import timedelta
8from decimal import Decimal
9from typing import TYPE_CHECKING
10from typing import Any
11from typing import Literal
12from typing import TypedDict
14import magic
15from django.conf import settings
16from django.contrib.auth.models import Group
17from django.contrib.auth.models import User
18from django.contrib.contenttypes.models import ContentType
19from django.core.exceptions import ValidationError
20from django.core.validators import DecimalValidator
21from django.core.validators import EmailValidator
22from django.core.validators import MaxLengthValidator
23from django.core.validators import MaxValueValidator
24from django.core.validators import MinValueValidator
25from django.core.validators import RegexValidator
26from django.core.validators import integer_validator
27from django.db.models import Count
28from django.db.models import Q
29from django.db.models.functions import Lower
30from django.utils import timezone
31from django.utils.crypto import get_random_string
32from django.utils.dateparse import parse_datetime
33from django.utils.text import slugify
34from django.utils.timezone import get_current_timezone
35from django.utils.timezone import is_naive
36from django.utils.timezone import make_aware
37from django.utils.translation import gettext as _
38from drf_spectacular.utils import extend_schema_field
39from drf_spectacular.utils import extend_schema_serializer
40from drf_writable_nested.serializers import NestedUpdateMixin
41from guardian.core import ObjectPermissionChecker
42from guardian.shortcuts import get_users_with_perms
43from guardian.utils import get_group_obj_perms_model
44from guardian.utils import get_user_obj_perms_model
45from rest_framework import fields
46from rest_framework import serializers
47from rest_framework.exceptions import PermissionDenied
48from rest_framework.fields import SerializerMethodField
49from rest_framework.filters import OrderingFilter
50from rest_framework.utils import model_meta
52if settings.AUDIT_LOG_ENABLED: 52 ↛ 56line 52 didn't jump to line 56 because the condition on line 52 was always true
53 from auditlog.context import set_actor
56from documents import bulk_edit
57from documents.data_models import DocumentSource
58from documents.filters import CustomFieldQueryParser
59from documents.models import Correspondent
60from documents.models import CustomField
61from documents.models import CustomFieldInstance
62from documents.models import Document
63from documents.models import DocumentBarcode
64from documents.models import DocumentType
65from documents.models import MatchingModel
66from documents.models import Note
67from documents.models import PaperlessTask
68from documents.models import SavedView
69from documents.models import SavedViewFilterRule
70from documents.models import ShareLink
71from documents.models import ShareLinkBundle
72from documents.models import StoragePath
73from documents.models import Tag
74from documents.models import UiSettings
75from documents.models import Workflow
76from documents.models import WorkflowAction
77from documents.models import WorkflowActionEmail
78from documents.models import WorkflowActionWebhook
79from documents.models import WorkflowTrigger
80from documents.parsers import is_mime_type_supported
81from documents.permissions import get_document_count_filter_for_user
82from documents.permissions import get_groups_with_only_permission
83from documents.permissions import has_perms_owner_aware
84from documents.permissions import permitted_document_ids
85from documents.permissions import restrict_queryset_to_visible
86from documents.permissions import set_permissions_for_object
87from documents.regex import validate_regex_pattern
88from documents.templating.filepath import validate_filepath_template_and_render
89from documents.templating.utils import convert_format_str_to_template_format
90from documents.templating.workflows import validate_workflow_template
91from documents.validators import uri_validator
92from documents.validators import url_validator
93from documents.versioning import has_prefetched_effective_content
94from documents.versioning import sort_versions_newest_first
96if TYPE_CHECKING: 96 ↛ 97line 96 didn't jump to line 97 because the condition on line 96 was never true
97 from collections.abc import Iterable
99 from django.db.models.query import QuerySet
100 from rest_framework.relations import ManyRelatedField
101 from rest_framework.relations import RelatedField
104logger = logging.getLogger("paperless.serializers")
107# https://www.django-rest-framework.org/api-guide/serializers/#example
108class DynamicFieldsModelSerializer(serializers.ModelSerializer[Any]):
109 """
110 A ModelSerializer that takes an additional `fields` argument that
111 controls which fields should be displayed.
112 """
114 def __init__(self, *args, **kwargs) -> None:
115 # Don't pass the 'fields' arg up to the superclass
116 fields = kwargs.pop("fields", None)
118 # Instantiate the superclass normally
119 super().__init__(*args, **kwargs)
121 if fields is not None:
122 # Drop any fields that are not specified in the `fields` argument.
123 allowed = set(fields)
124 existing = set(self.fields)
125 for field_name in existing - allowed:
126 self.fields.pop(field_name)
129class DocumentUpdateFieldsModelSerializer(DynamicFieldsModelSerializer):
130 stale_update_excluded_fields = frozenset({"filename", "archive_filename"})
132 def _get_update_fields(self, validated_data) -> list[str]:
133 model_fields = {
134 field.name
135 for field in self.Meta.model._meta.concrete_fields
136 if field.name not in self.stale_update_excluded_fields
137 }
138 update_fields = [
139 field_name for field_name in validated_data if field_name in model_fields
140 ]
141 if "modified" in model_fields and "modified" not in update_fields:
142 update_fields.append("modified")
143 return update_fields
145 def update(self, instance, validated_data):
146 serializers.raise_errors_on_nested_writes("update", self, validated_data)
147 info = model_meta.get_field_info(instance)
149 m2m_fields = []
150 for attr, value in validated_data.items():
151 if attr in info.relations and info.relations[attr].to_many:
152 m2m_fields.append((attr, value))
153 else:
154 setattr(instance, attr, value)
156 # File names are managed by post-save file handling. Saving only the
157 # serializer-updated fields prevents stale in-memory path values from
158 # overwriting a concurrent move.
159 instance.save(update_fields=self._get_update_fields(validated_data))
161 for attr, value in m2m_fields:
162 field = getattr(instance, attr)
163 field.set(value)
165 return instance
168class MatchingModelSerializer(serializers.ModelSerializer[Any]):
169 document_count = serializers.IntegerField(read_only=True)
171 def get_slug(self, obj) -> str:
172 return slugify(obj.name)
174 slug = SerializerMethodField()
176 def validate(self, data):
177 # TODO: remove pending https://github.com/encode/django-rest-framework/issues/7173
178 name = data.get(
179 "name",
180 self.instance.name if hasattr(self.instance, "name") else None,
181 )
182 owner = (
183 data["owner"]
184 if "owner" in data
185 else self.user
186 if hasattr(self, "user")
187 else None
188 )
189 pk = self.instance.pk if hasattr(self.instance, "pk") else None
190 if ("name" in data or "owner" in data) and self.Meta.model.objects.filter(
191 name=name,
192 owner=owner,
193 ).exclude(pk=pk).exists():
194 raise serializers.ValidationError(
195 {"error": "Object violates owner / name unique constraint"},
196 )
197 return data
199 def validate_match(self, match):
200 if (
201 "matching_algorithm" in self.initial_data
202 and self.initial_data["matching_algorithm"] == MatchingModel.MATCH_REGEX
203 ):
204 try:
205 validate_regex_pattern(match)
206 except ValueError as e:
207 logger.debug(f"Invalid regular expression: {e!s}")
208 raise serializers.ValidationError(
209 "Invalid regular expression, see log for details.",
210 )
211 return match
214PERMISSION_ACTIONS = ("view", "change")
217class SetPermissionsMixin:
218 def _validate_user_ids(self, user_ids):
219 users = User.objects.none()
220 if user_ids is not None:
221 users = User.objects.filter(id__in=user_ids)
222 if not users.count() == len(user_ids):
223 raise serializers.ValidationError(
224 "Some users in don't exist or were specified twice.",
225 )
226 return users
228 def _validate_group_ids(self, group_ids):
229 groups = Group.objects.none()
230 if group_ids is not None:
231 groups = Group.objects.filter(id__in=group_ids)
232 if not groups.count() == len(group_ids):
233 raise serializers.ValidationError(
234 "Some groups in don't exist or were specified twice.",
235 )
236 return groups
238 def validate_set_permissions(self, set_permissions=None):
239 permissions_dict = {action: {} for action in PERMISSION_ACTIONS}
240 if set_permissions is not None: 240 ↛ 255line 240 didn't jump to line 255 because the condition on line 240 was always true
241 for action in PERMISSION_ACTIONS:
242 if action in set_permissions:
243 if "users" in set_permissions[action]:
244 users = set_permissions[action]["users"]
245 permissions_dict[action]["users"] = self._validate_user_ids(
246 users,
247 )
248 if "groups" in set_permissions[action]:
249 groups = set_permissions[action]["groups"]
250 permissions_dict[action]["groups"] = self._validate_group_ids(
251 groups,
252 )
253 else:
254 del permissions_dict[action]
255 return permissions_dict
257 def _set_permissions(self, permissions, object) -> None:
258 set_permissions_for_object(permissions, object)
261class SerializerWithPerms(serializers.Serializer[dict[str, Any]]):
262 def __init__(self, *args, **kwargs) -> None:
263 self.user = kwargs.pop("user", None)
264 self.full_perms = kwargs.pop("full_perms", False)
265 self.all_fields = kwargs.pop("all_fields", False)
266 super().__init__(*args, **kwargs)
269class PermissionSetSerializer(serializers.Serializer[dict[str, Any]]):
270 users = serializers.ListField(
271 child=serializers.IntegerField(),
272 required=False,
273 allow_null=True,
274 )
275 groups = serializers.ListField(
276 child=serializers.IntegerField(),
277 required=False,
278 allow_null=True,
279 )
282class SetPermissionsSerializer(serializers.Serializer[dict[str, Any]]):
283 view = PermissionSetSerializer(required=False)
284 change = PermissionSetSerializer(required=False)
286 def to_internal_value(self, data):
287 if isinstance(data, dict):
288 unknown_keys = set(data) - set(PERMISSION_ACTIONS)
289 if unknown_keys:
290 raise serializers.ValidationError(
291 {key: "Unknown permission action." for key in sorted(unknown_keys)},
292 )
293 return super().to_internal_value(data)
296class OwnedObjectSerializer(
297 SerializerWithPerms,
298 serializers.ModelSerializer[Any],
299 SetPermissionsMixin,
300):
301 def __init__(self, *args, **kwargs) -> None:
302 super().__init__(*args, **kwargs)
304 if not self.all_fields:
305 try:
306 if self.full_perms:
307 self.fields.pop("user_can_change")
308 self.fields.pop("is_shared_by_requester")
309 else:
310 self.fields.pop("permissions")
311 except KeyError:
312 pass
314 def _get_perms(self, obj, codename: str, target: Literal["users", "groups"]):
315 """
316 Get the given permissions from context or from django-guardian.
318 :param codename: The permission codename, e.g. 'view' or 'change'
319 :param target: 'users' or 'groups'
320 """
321 key = f"{target}_{codename}_perms"
322 cached = self.context.get(key, {}).get(obj.pk)
323 if cached is not None:
324 return list(cached)
326 # Permission not found in the context, get it from guardian
327 if target == "users":
328 return list(
329 get_users_with_perms(
330 obj,
331 only_with_perms_in=[f"{codename}_{obj.__class__.__name__.lower()}"],
332 with_group_users=False,
333 ).values_list("id", flat=True),
334 )
335 else: # groups
336 return list(
337 get_groups_with_only_permission(
338 obj,
339 codename=f"{codename}_{obj.__class__.__name__.lower()}",
340 ).values_list("id", flat=True),
341 )
343 @extend_schema_field(
344 field={
345 "type": "object",
346 "properties": {
347 "view": {
348 "type": "object",
349 "properties": {
350 "users": {
351 "type": "array",
352 "items": {"type": "integer"},
353 },
354 "groups": {
355 "type": "array",
356 "items": {"type": "integer"},
357 },
358 },
359 },
360 "change": {
361 "type": "object",
362 "properties": {
363 "users": {
364 "type": "array",
365 "items": {"type": "integer"},
366 },
367 "groups": {
368 "type": "array",
369 "items": {"type": "integer"},
370 },
371 },
372 },
373 },
374 },
375 )
376 def get_permissions(self, obj) -> dict:
377 return {
378 "view": {
379 "users": self._get_perms(obj, "view", "users"),
380 "groups": self._get_perms(obj, "view", "groups"),
381 },
382 "change": {
383 "users": self._get_perms(obj, "change", "users"),
384 "groups": self._get_perms(obj, "change", "groups"),
385 },
386 }
388 def get_user_can_change(self, obj) -> bool:
389 if obj.owner is None or obj.owner == self.user:
390 return True
391 if self.user is None:
392 return False
393 if self.user.is_active and self.user.is_superuser: 393 ↛ 402line 393 didn't jump to line 402 because the condition on line 393 was always true
394 # Mirrors guardian's own ObjectPermissionChecker.has_perm() shortcut --
395 # superusers aren't necessarily granted explicit object permissions,
396 # so the batched context below would otherwise incorrectly say no.
397 return True
399 # Prefer the page-level batch computed by BulkPermissionMixin
400 # (get_serializer_context) over a fresh per-object guardian check,
401 # which would otherwise query the permission tables once per row.
402 users_change_perms = self.context.get("users_change_perms")
403 groups_change_perms = self.context.get("groups_change_perms")
404 if users_change_perms is not None and groups_change_perms is not None:
405 if self.user.pk in users_change_perms.get(obj.pk, []):
406 return True
407 user_group_ids = getattr(self, "_user_group_ids", None)
408 if user_group_ids is None:
409 user_group_ids = set(self.user.groups.values_list("id", flat=True))
410 self._user_group_ids = user_group_ids
411 return bool(
412 user_group_ids.intersection(groups_change_perms.get(obj.pk, [])),
413 )
415 checker = ObjectPermissionChecker(self.user)
416 return checker.has_perm(f"change_{obj.__class__.__name__.lower()}", obj)
418 @staticmethod
419 def get_shared_object_pks(objects: Iterable):
420 """
421 Return the primary keys of the subset of objects that are shared.
422 """
423 try:
424 first_obj = next(iter(objects))
425 except StopIteration:
426 return set()
428 ctype = ContentType.objects.get_for_model(first_obj)
429 object_pks = list(obj.pk for obj in objects)
430 pk_type = type(first_obj.pk)
432 def get_pks_for_permission_type(model):
433 return map(
434 pk_type, # coerce the pk to be the same type of the provided objects
435 model.objects.filter(
436 content_type=ctype,
437 object_pk__in=object_pks,
438 )
439 .values_list("object_pk", flat=True)
440 .distinct(),
441 )
443 UserObjectPermission = get_user_obj_perms_model()
444 GroupObjectPermission = get_group_obj_perms_model()
445 user_permission_pks = get_pks_for_permission_type(UserObjectPermission)
446 group_permission_pks = get_pks_for_permission_type(GroupObjectPermission)
448 return set(user_permission_pks) | set(group_permission_pks)
450 def get_is_shared_by_requester(self, obj: Document) -> bool:
451 # First check the context to see if `shared_object_pks` is set by the parent.
452 shared_object_pks = self.context.get("shared_object_pks")
453 # If not just check if the current object is shared.
454 if shared_object_pks is None:
455 shared_object_pks = self.get_shared_object_pks([obj])
456 return obj.owner == self.user and obj.id in shared_object_pks
458 permissions = SerializerMethodField(read_only=True, required=False)
459 user_can_change = SerializerMethodField(read_only=True, required=False)
460 is_shared_by_requester = SerializerMethodField(read_only=True, required=False)
462 set_permissions = SetPermissionsSerializer(
463 label="Set permissions",
464 required=False,
465 write_only=True,
466 )
467 # other methods in mixin
469 def validate_unique_together(self, validated_data, instance=None) -> None:
470 # workaround for https://github.com/encode/django-rest-framework/issues/9358
471 if "owner" in validated_data and "name" in self.Meta.fields:
472 name = validated_data.get("name", instance.name if instance else None)
473 objects = (
474 self.Meta.model.objects.exclude(pk=instance.pk)
475 if instance
476 else self.Meta.model.objects.all()
477 )
478 not_unique = objects.filter(
479 owner=validated_data["owner"],
480 name=name,
481 ).exists()
482 if not_unique:
483 raise serializers.ValidationError(
484 {"error": "Object violates owner / name unique constraint"},
485 )
487 def create(self, validated_data):
488 # default to current user if not set
489 request = self.context.get("request")
490 if (
491 "owner" not in validated_data
492 or (request is not None and "owner" not in request.data)
493 ) and self.user:
494 validated_data["owner"] = self.user
495 permissions = None
496 if "set_permissions" in validated_data:
497 permissions = validated_data.pop("set_permissions")
498 self.validate_unique_together(validated_data)
499 instance = super().create(validated_data)
500 if permissions is not None:
501 self._set_permissions(permissions, instance)
502 return instance
504 def update(self, instance, validated_data):
505 user = getattr(self, "user", None)
506 is_superuser = user.is_superuser if user is not None else False
507 is_owner = instance.owner == user if user is not None else False
508 is_unowned = instance.owner is None
510 if ( 510 ↛ 514line 510 didn't jump to line 514 because the condition on line 510 was never true
511 ("owner" in validated_data and validated_data["owner"] != instance.owner)
512 or "set_permissions" in validated_data
513 ) and not (is_superuser or is_owner or is_unowned):
514 raise PermissionDenied(
515 _("Insufficient permissions."),
516 )
518 if "set_permissions" in validated_data:
519 self._set_permissions(validated_data["set_permissions"], instance)
520 self.validate_unique_together(validated_data, instance)
521 return super().update(instance, validated_data)
524class OwnedObjectListSerializer(serializers.ListSerializer[Any]):
525 def to_representation(self, documents):
526 self.child.context["shared_object_pks"] = self.child.get_shared_object_pks(
527 documents,
528 )
529 return super().to_representation(documents)
532class CorrespondentSerializer(MatchingModelSerializer, OwnedObjectSerializer):
533 last_correspondence = serializers.DateField(read_only=True, required=False)
535 class Meta:
536 model = Correspondent
537 fields = (
538 "id",
539 "slug",
540 "name",
541 "match",
542 "matching_algorithm",
543 "is_insensitive",
544 "document_count",
545 "last_correspondence",
546 "owner",
547 "permissions",
548 "user_can_change",
549 "set_permissions",
550 )
553class DocumentTypeSerializer(MatchingModelSerializer, OwnedObjectSerializer):
554 class Meta:
555 model = DocumentType
556 fields = (
557 "id",
558 "slug",
559 "name",
560 "match",
561 "matching_algorithm",
562 "is_insensitive",
563 "document_count",
564 "owner",
565 "permissions",
566 "user_can_change",
567 "set_permissions",
568 )
571class DeprecatedColors:
572 COLOURS = (
573 (1, "#a6cee3"),
574 (2, "#1f78b4"),
575 (3, "#b2df8a"),
576 (4, "#33a02c"),
577 (5, "#fb9a99"),
578 (6, "#e31a1c"),
579 (7, "#fdbf6f"),
580 (8, "#ff7f00"),
581 (9, "#cab2d6"),
582 (10, "#6a3d9a"),
583 (11, "#b15928"),
584 (12, "#000000"),
585 (13, "#cccccc"),
586 )
589@extend_schema_field(
590 serializers.ChoiceField(
591 choices=DeprecatedColors.COLOURS,
592 ),
593)
594class ColorField(serializers.Field):
595 def to_internal_value(self, data):
596 for id, color in DeprecatedColors.COLOURS:
597 if id == data:
598 return color
599 raise serializers.ValidationError
601 def to_representation(self, value):
602 for id, color in DeprecatedColors.COLOURS:
603 if color == value:
604 return id
605 return 1
608class TagSerializer(MatchingModelSerializer, OwnedObjectSerializer):
609 def get_text_color(self, obj) -> str:
610 try:
611 h = obj.color.lstrip("#")
612 rgb = tuple(int(h[i : i + 2], 16) / 256 for i in (0, 2, 4))
613 luminance = math.sqrt(
614 0.299 * math.pow(rgb[0], 2)
615 + 0.587 * math.pow(rgb[1], 2)
616 + 0.114 * math.pow(rgb[2], 2),
617 )
618 return "#ffffff" if luminance < 0.53 else "#000000"
619 except ValueError:
620 return "#000000"
622 text_color = serializers.SerializerMethodField()
624 # map to treenode's tn_parent
625 parent = serializers.PrimaryKeyRelatedField(
626 queryset=Tag.objects.all(),
627 allow_null=True,
628 required=False,
629 source="tn_parent",
630 )
632 @extend_schema_field(
633 field=serializers.ListSerializer(
634 child=serializers.PrimaryKeyRelatedField(
635 queryset=Tag.objects.all(),
636 ),
637 ),
638 )
639 def get_children(self, obj):
640 children_map = self.context.get("children_map")
641 if children_map is not None:
642 children = children_map.get(obj.pk, [])
643 else:
644 filter_q = self.context.get("document_count_filter")
645 request = self.context.get("request")
646 if filter_q is None:
647 user = getattr(request, "user", None) if request else None
648 filter_q = get_document_count_filter_for_user(user)
649 self.context["document_count_filter"] = filter_q
651 children = (
652 obj.get_children_queryset()
653 .select_related("owner")
654 .annotate(document_count=Count("documents", filter=filter_q))
655 )
656 user = getattr(request, "user", None) if request else self.user
657 children = restrict_queryset_to_visible(children, user, "view_tag")
659 view = self.context.get("view")
660 ordering = (
661 OrderingFilter().get_ordering(request, children, view)
662 if request and view
663 else None
664 )
665 ordering = ordering or (Lower("name"),)
666 children = children.order_by(*ordering)
668 if not children:
669 return []
671 serializer = TagSerializer(
672 children,
673 many=True,
674 user=self.user,
675 full_perms=self.full_perms,
676 all_fields=self.all_fields,
677 context=self.context,
678 )
679 return serializer.data
681 # children as nested Tag objects
682 children = serializers.SerializerMethodField()
684 class Meta:
685 model = Tag
686 fields = (
687 "id",
688 "slug",
689 "name",
690 "color",
691 "text_color",
692 "match",
693 "matching_algorithm",
694 "is_insensitive",
695 "is_inbox_tag",
696 "document_count",
697 "owner",
698 "permissions",
699 "user_can_change",
700 "set_permissions",
701 "parent",
702 "children",
703 )
705 def validate_color(self, color):
706 regex = r"#[0-9a-fA-F]{6}"
707 if not re.match(regex, color):
708 raise serializers.ValidationError(_("Invalid color."))
709 return color
711 def validate(self, attrs):
712 # Validate when changing parent
713 parent = attrs.get(
714 "tn_parent",
715 self.instance.get_parent() if self.instance else None,
716 )
718 if self.instance:
719 # Temporarily set parent on the instance if updating and use model clean()
720 original_parent = self.instance.get_parent()
721 try:
722 # Temporarily set tn_parent in-memory to validate clean()
723 self.instance.tn_parent = parent
724 self.instance.clean()
725 except ValidationError as e:
726 logger.debug("Tag parent validation failed: %s", e)
727 raise e
728 finally:
729 self.instance.tn_parent = original_parent
730 else:
731 # For new instances, create a transient Tag and validate
732 temp = Tag(tn_parent=parent)
733 try:
734 temp.clean()
735 except ValidationError as e:
736 logger.debug("Tag parent validation failed: %s", e)
737 raise e
739 return super().validate(attrs)
742class CorrespondentField(serializers.PrimaryKeyRelatedField[Correspondent]):
743 def get_queryset(self):
744 return Correspondent.objects.all()
747class TagsField(serializers.PrimaryKeyRelatedField[Tag]):
748 def get_queryset(self):
749 return Tag.objects.all()
752class DocumentTypeField(serializers.PrimaryKeyRelatedField[DocumentType]):
753 def get_queryset(self):
754 return DocumentType.objects.all()
757class StoragePathField(serializers.PrimaryKeyRelatedField[StoragePath]):
758 def get_queryset(self):
759 return StoragePath.objects.all()
762class CustomFieldSerializer(serializers.ModelSerializer[CustomField]):
763 data_type = serializers.ChoiceField(
764 choices=CustomField.FieldDataType,
765 read_only=False,
766 )
768 document_count = serializers.IntegerField(read_only=True)
770 class Meta:
771 model = CustomField
772 fields = [
773 "id",
774 "name",
775 "data_type",
776 "extra_data",
777 "document_count",
778 ]
780 def validate(self, attrs):
781 # TODO: remove pending https://github.com/encode/django-rest-framework/issues/7173
782 name = attrs.get(
783 "name",
784 self.instance.name if hasattr(self.instance, "name") else None,
785 )
786 objects = (
787 self.Meta.model.objects.exclude(
788 pk=self.instance.pk,
789 )
790 if self.instance is not None
791 else self.Meta.model.objects.all()
792 )
793 if ("name" in attrs) and objects.filter( 793 ↛ 796line 793 didn't jump to line 796 because the condition on line 793 was never true
794 name=name,
795 ).exists():
796 raise serializers.ValidationError(
797 {"error": "Object violates name unique constraint"},
798 )
799 if (
800 "data_type" in attrs
801 and attrs["data_type"] == CustomField.FieldDataType.SELECT
802 ) or (
803 self.instance
804 and self.instance.data_type == CustomField.FieldDataType.SELECT
805 ):
806 if ( 806 ↛ 820line 806 didn't jump to line 820 because the condition on line 806 was always true
807 "extra_data" not in attrs
808 or "select_options" not in attrs["extra_data"]
809 or not isinstance(attrs["extra_data"]["select_options"], list)
810 or len(attrs["extra_data"]["select_options"]) == 0
811 or not all(
812 len(option.get("label", "")) > 0
813 for option in attrs["extra_data"]["select_options"]
814 )
815 ):
816 raise serializers.ValidationError(
817 {"error": "extra_data.select_options must be a valid list"},
818 )
819 # labels are valid, generate ids if not present
820 for option in attrs["extra_data"]["select_options"]:
821 if option.get("id") is None:
822 option["id"] = get_random_string(length=16)
823 elif ( 823 ↛ 837line 823 didn't jump to line 837 because the condition on line 823 was never true
824 "data_type" in attrs
825 and attrs["data_type"] == CustomField.FieldDataType.MONETARY
826 and "extra_data" in attrs
827 and "default_currency" in attrs["extra_data"]
828 and attrs["extra_data"]["default_currency"] is not None
829 and (
830 not isinstance(attrs["extra_data"]["default_currency"], str)
831 or (
832 len(attrs["extra_data"]["default_currency"]) > 0
833 and len(attrs["extra_data"]["default_currency"]) != 3
834 )
835 )
836 ):
837 raise serializers.ValidationError(
838 {"error": "extra_data.default_currency must be a 3-character string"},
839 )
840 return super().validate(attrs)
843class ReadWriteSerializerMethodField(serializers.SerializerMethodField):
844 """
845 Based on https://stackoverflow.com/a/62579804
846 """
848 def __init__(self, method_name=None, *args, **kwargs) -> None:
849 self.method_name = method_name
850 kwargs["source"] = "*"
851 super(serializers.SerializerMethodField, self).__init__(*args, **kwargs)
853 def to_internal_value(self, data):
854 return {self.field_name: data}
857def validate_documentlink_targets(user, doc_ids):
858 if Document.objects.filter(id__in=doc_ids).count() != len(doc_ids):
859 raise serializers.ValidationError(
860 "Some documents in value don't exist or were specified twice.",
861 )
863 if user is None:
864 return
866 if (
867 Document.objects.filter(id__in=doc_ids)
868 .exclude(id__in=permitted_document_ids(user, perm="change_document"))
869 .exists()
870 ):
871 raise PermissionDenied(
872 _("Insufficient permissions."),
873 )
876class CustomFieldInstanceSerializer(serializers.ModelSerializer[CustomFieldInstance]):
877 field = serializers.PrimaryKeyRelatedField(queryset=CustomField.objects.all())
878 value = ReadWriteSerializerMethodField(allow_null=True)
880 def create(self, validated_data):
881 # An instance is attached to a document
882 document: Document = validated_data["document"]
883 # And to a CustomField
884 custom_field: CustomField = validated_data["field"]
885 # This key must exist, as it is validated
886 data_store_name = CustomFieldInstance.get_value_field_name(
887 custom_field.data_type,
888 )
890 if custom_field.data_type == CustomField.FieldDataType.DOCUMENTLINK:
891 # prior to update so we can look for any docs that are going to be removed
892 bulk_edit.reflect_doclinks(document, custom_field, validated_data["value"])
894 # Actually update or create the instance, providing the value
895 # to fill in the correct attribute based on the type
896 instance, _ = CustomFieldInstance.objects.update_or_create(
897 document=document,
898 field=custom_field,
899 defaults={data_store_name: validated_data["value"]},
900 )
901 return instance
903 def get_value(self, obj: CustomFieldInstance) -> str | int | float | dict | None:
904 return obj.value
906 def validate(self, data):
907 """
908 Probably because we're kind of doing it odd, validation from the model
909 doesn't run against the field "value", so we have to re-create it here.
911 Don't like it, but it is better than returning an HTTP 500 when the database
912 hates the value
913 """
914 data = super().validate(data)
915 field: CustomField = data["field"]
916 if "value" in data and data["value"] is not None:
917 if (
918 field.data_type == CustomField.FieldDataType.URL
919 and len(data["value"]) > 0
920 ):
921 uri_validator(data["value"])
922 elif field.data_type == CustomField.FieldDataType.INT:
923 integer_validator(data["value"])
924 try:
925 value_int = int(data["value"])
926 except (TypeError, ValueError):
927 raise serializers.ValidationError("Enter a valid integer.")
928 # Keep values within the PostgreSQL integer range
929 MinValueValidator(-2147483648)(value_int)
930 MaxValueValidator(2147483647)(value_int)
931 elif (
932 field.data_type == CustomField.FieldDataType.MONETARY
933 and data["value"] != ""
934 ):
935 try:
936 # First try to validate as a number from legacy format
937 DecimalValidator(max_digits=12, decimal_places=2)(
938 Decimal(str(data["value"])),
939 )
940 except Exception:
941 # If that fails, try to validate as a monetary string
942 RegexValidator(
943 regex=r"^[A-Z]{3}-?\d+(\.\d{1,2})$",
944 message="Must be a two-decimal number with optional currency code e.g. GBP123.45",
945 )(data["value"])
946 elif field.data_type == CustomField.FieldDataType.STRING:
947 MaxLengthValidator(limit_value=128)(data["value"])
948 elif field.data_type == CustomField.FieldDataType.SELECT:
949 select_options = field.extra_data["select_options"]
950 try:
951 next(
952 option
953 for option in select_options
954 if option["id"] == data["value"]
955 )
956 except Exception:
957 raise serializers.ValidationError(
958 f"Value must be an id of an element in {select_options}",
959 )
960 elif field.data_type == CustomField.FieldDataType.DOCUMENTLINK:
961 if not (isinstance(data["value"], list) or data["value"] is None):
962 raise serializers.ValidationError(
963 "Value must be a list",
964 )
965 doc_ids = data["value"]
966 request = self.context.get("request")
967 validate_documentlink_targets(
968 getattr(request, "user", None) if request is not None else None,
969 doc_ids,
970 )
971 elif field.data_type == CustomField.FieldDataType.DATE:
972 data["value"] = serializers.DateField().to_internal_value(data["value"])
974 return data
976 class Meta:
977 model = CustomFieldInstance
978 fields = [
979 "value",
980 "field",
981 ]
984class BasicUserSerializer(serializers.ModelSerializer[User]):
985 # Different than paperless.serializers.UserSerializer
986 class Meta:
987 model = User
988 fields = ["id", "username", "first_name", "last_name"]
991class DocumentBarcodeSerializer(serializers.ModelSerializer[DocumentBarcode]):
992 class Meta:
993 model = DocumentBarcode
994 fields = ["page", "value", "format"]
997class NotesSerializer(serializers.ModelSerializer[Note]):
998 user = BasicUserSerializer(read_only=True)
1000 class Meta:
1001 model = Note
1002 fields = ["id", "note", "created", "user"]
1003 ordering = ["-created"]
1006def _get_viewable_duplicates(
1007 document: Document,
1008 user: User | None,
1009) -> QuerySet[Document]:
1010 checksums = {document.checksum}
1011 if document.archive_checksum:
1012 checksums.add(document.archive_checksum)
1013 duplicates = Document.global_objects.filter(
1014 Q(checksum__in=checksums) | Q(archive_checksum__in=checksums),
1015 ).exclude(pk=document.pk)
1016 duplicates = duplicates.filter(root_document__isnull=True)
1017 duplicates = duplicates.order_by("-created")
1018 allowed_ids = permitted_document_ids(user, include_deleted=True)
1019 return duplicates.filter(id__in=allowed_ids)
1022class DuplicateDocumentSummarySerializer(serializers.Serializer[dict[str, Any]]):
1023 id = serializers.IntegerField()
1024 title = serializers.CharField()
1025 deleted_at = serializers.DateTimeField(allow_null=True)
1028class _DocumentVersionInfo(TypedDict):
1029 id: int
1030 added: datetime
1031 version_label: str | None
1032 checksum: str | None
1033 is_root: bool
1036class DocumentVersionInfoSerializer(serializers.Serializer[_DocumentVersionInfo]):
1037 id = serializers.IntegerField()
1038 added = serializers.DateTimeField()
1039 version_label = serializers.CharField(required=False, allow_null=True)
1040 checksum = serializers.CharField(required=False, allow_null=True)
1041 is_root = serializers.BooleanField()
1044@extend_schema_serializer(
1045 deprecate_fields=["created_date"],
1046)
1047class DocumentSerializer(
1048 OwnedObjectSerializer,
1049 NestedUpdateMixin,
1050 DocumentUpdateFieldsModelSerializer,
1051):
1052 correspondent = CorrespondentField(allow_null=True)
1053 tags = TagsField(many=True)
1054 document_type = DocumentTypeField(allow_null=True)
1055 storage_path = StoragePathField(allow_null=True)
1057 original_file_name = SerializerMethodField()
1058 archived_file_name = SerializerMethodField()
1059 created_date = serializers.DateField(required=False)
1060 page_count = SerializerMethodField()
1061 duplicate_documents = SerializerMethodField()
1063 notes = NotesSerializer(many=True, required=False, read_only=True)
1064 root_document: RelatedField[Document, Document, Any] | ManyRelatedField = (
1065 serializers.PrimaryKeyRelatedField(read_only=True)
1066 )
1067 versions = SerializerMethodField()
1069 custom_fields = CustomFieldInstanceSerializer(
1070 many=True,
1071 allow_null=False,
1072 required=False,
1073 )
1075 owner = serializers.PrimaryKeyRelatedField(
1076 queryset=User.objects.all(),
1077 required=False,
1078 allow_null=True,
1079 )
1081 remove_inbox_tags = serializers.BooleanField(
1082 default=False,
1083 write_only=True,
1084 allow_null=True,
1085 required=False,
1086 )
1088 def get_page_count(self, obj) -> int | None:
1089 # Like content versions get their own page count from the newest version,
1090 # use the prefetched versions cache to avoid an extra query
1091 prefetched_cache = getattr(obj, "_prefetched_objects_cache", None)
1092 prefetched_versions = (
1093 prefetched_cache.get("versions")
1094 if isinstance(prefetched_cache, dict)
1095 else None
1096 )
1097 if obj.root_document_id is None and prefetched_versions:
1098 return sort_versions_newest_first(prefetched_versions)[0].page_count
1099 return obj.page_count
1101 @extend_schema_field(DuplicateDocumentSummarySerializer(many=True))
1102 def get_duplicate_documents(self, obj):
1103 view = self.context.get("view")
1104 if view and getattr(view, "action", None) != "retrieve":
1105 return []
1106 request = self.context.get("request")
1107 user = request.user if request else None
1108 duplicates = _get_viewable_duplicates(obj, user)
1109 return list(duplicates.values("id", "title", "deleted_at"))
1111 @extend_schema_field(DocumentVersionInfoSerializer(many=True))
1112 def get_versions(self, obj):
1113 root_doc = obj if obj.root_document_id is None else obj.root_document
1114 if root_doc is None:
1115 return []
1117 prefetched_cache = getattr(obj, "_prefetched_objects_cache", None)
1118 prefetched_versions = (
1119 prefetched_cache.get("versions")
1120 if isinstance(prefetched_cache, dict)
1121 else None
1122 )
1124 versions: list[Document]
1125 if prefetched_versions is not None:
1126 versions = [*prefetched_versions, root_doc]
1127 else:
1128 versions_qs = Document.objects.filter(root_document=root_doc).only(
1129 "id",
1130 "added",
1131 "checksum",
1132 "version_label",
1133 "root_document_id",
1134 "version_index",
1135 )
1136 versions = [*versions_qs, root_doc]
1138 versions = sort_versions_newest_first(versions)
1140 def build_info(doc: Document) -> _DocumentVersionInfo:
1141 return {
1142 "id": doc.id,
1143 "added": doc.added,
1144 "version_label": doc.version_label,
1145 "checksum": doc.checksum,
1146 "is_root": doc.id == root_doc.id,
1147 }
1149 return [build_info(doc) for doc in versions]
1151 def get_original_file_name(self, obj) -> str | None:
1152 return obj.original_filename
1154 def get_archived_file_name(self, obj) -> str | None:
1155 if obj.has_archive_version:
1156 return obj.get_public_filename(archive=True)
1157 else:
1158 return None
1160 def to_representation(self, instance):
1161 doc = super().to_representation(instance)
1162 if "content" in self.fields and has_prefetched_effective_content(instance):
1163 # Only resolve version-aware content when it's cheap: an SQL
1164 # annotation or a versions prefetch is already on the instance.
1165 # A caller that set up neither (e.g. TrashView, GlobalSearchView,
1166 # which build their own querysets) gets the document's own,
1167 # unresolved content instead of paying for an extra per-instance
1168 # query -- same as before effective_content resolution existed.
1169 doc["content"] = instance.get_effective_content() or ""
1170 if self.truncate_content and "content" in self.fields:
1171 doc["content"] = doc.get("content")[0:550]
1172 return doc
1174 def to_internal_value(self, data):
1175 if (
1176 "created" in data
1177 and isinstance(data["created"], str)
1178 and ":" in data["created"]
1179 ):
1180 # Handle old format of isoformat datetime string
1181 parsed = parse_datetime(data["created"])
1182 if parsed:
1183 if is_naive(parsed):
1184 parsed = make_aware(parsed, get_current_timezone())
1185 data["created"] = parsed.astimezone().date()
1186 return super().to_internal_value(data)
1188 def validate(self, attrs):
1189 if (
1190 "archive_serial_number" in attrs
1191 and attrs["archive_serial_number"] is not None
1192 and len(str(attrs["archive_serial_number"])) > 0
1193 and Document.deleted_objects.filter(
1194 archive_serial_number=attrs["archive_serial_number"],
1195 ).exists()
1196 ):
1197 raise serializers.ValidationError(
1198 {
1199 "archive_serial_number": [
1200 "Document with this Archive Serial Number already exists in the trash.",
1201 ],
1202 },
1203 )
1204 return super().validate(attrs)
1206 def update(self, instance: Document, validated_data):
1207 if "created_date" in validated_data:
1208 if "created" not in validated_data:
1209 validated_data["created"] = validated_data["created_date"]
1210 logger.warning(
1211 "created_date is deprecated, use created instead",
1212 )
1213 validated_data.pop("created_date")
1214 if instance.custom_fields.count() > 0 and "custom_fields" in validated_data:
1215 incoming_custom_fields = [
1216 field["field"] for field in validated_data["custom_fields"]
1217 ]
1218 for custom_field_instance in instance.custom_fields.filter(
1219 field__data_type=CustomField.FieldDataType.DOCUMENTLINK,
1220 ):
1221 if (
1222 custom_field_instance.field not in incoming_custom_fields
1223 and custom_field_instance.value is not None
1224 ):
1225 # Doc link field is being removed entirely
1226 for doc_id in custom_field_instance.value:
1227 bulk_edit.remove_doclink(
1228 instance,
1229 custom_field_instance.field,
1230 doc_id,
1231 )
1232 if "tags" in validated_data:
1233 # Respect tag hierarchy on updates:
1234 # - Adding a child adds its ancestors
1235 # - Removing a parent removes all its descendants
1236 prev_tags = set(instance.tags.all())
1237 requested_tags = set(validated_data["tags"])
1239 # Tags newly added in this update and the ancestors they require
1240 added_tags = requested_tags - prev_tags
1241 required_by_add_tags = set(added_tags)
1242 for t in added_tags:
1243 required_by_add_tags.update(t.get_ancestors())
1245 # Tags being removed in this update and all descendants, except
1246 # those required by a tag that is being added in this same update
1247 removed_tags = prev_tags - requested_tags
1248 blocked_tags = set(removed_tags)
1249 for t in removed_tags:
1250 blocked_tags.update(t.get_descendants())
1251 blocked_tags.difference_update(required_by_add_tags)
1253 # Add all parent tags
1254 final_tags = set(requested_tags)
1255 for t in requested_tags:
1256 final_tags.update(t.get_ancestors())
1258 # Drop removed parents and their descendants
1259 final_tags.difference_update(blocked_tags)
1261 validated_data["tags"] = list(final_tags)
1262 if validated_data.get("remove_inbox_tags"):
1263 current_tag_ids = {t.pk for t in instance.tags.all()}
1264 tags = (
1265 validated_data["tags"]
1266 if "tags" in validated_data
1267 else list(instance.tags.all())
1268 )
1270 # Tags newly added in this update, plus their ancestors, are kept
1271 keep_ids: set[int] = set()
1272 for tag in tags:
1273 if tag.pk not in current_tag_ids:
1274 keep_ids.add(tag.pk)
1275 keep_ids.update(int(pk) for pk in tag.get_ancestors_pks())
1277 # Remove inbox tags and their descendants, except those being kept
1278 remove_ids: set[int] = set()
1279 for inbox_tag in (
1280 Tag.objects.filter(is_inbox_tag=True)
1281 .exclude(pk__in=keep_ids)
1282 .only("pk", "tn_descendants_pks")
1283 ):
1284 remove_ids.add(inbox_tag.pk)
1285 remove_ids.update(int(pk) for pk in inbox_tag.get_descendants_pks())
1287 validated_data["tags"] = [t for t in tags if t.pk not in remove_ids]
1289 if settings.AUDIT_LOG_ENABLED:
1290 with set_actor(self.user):
1291 super().update(instance, validated_data)
1292 else:
1293 super().update(instance, validated_data)
1295 # hard delete custom field instances that were soft deleted
1296 CustomFieldInstance.deleted_objects.filter(document=instance).delete()
1297 return instance
1299 def __init__(self, *args, **kwargs) -> None:
1300 self.truncate_content = kwargs.pop("truncate_content", False)
1302 # return full permissions if we're doing a PATCH or PUT
1303 context = kwargs.get("context")
1304 if context is not None and (
1305 context.get("request").method == "PATCH"
1306 or context.get("request").method == "PUT"
1307 ):
1308 kwargs["full_perms"] = True
1310 super().__init__(*args, **kwargs)
1312 class Meta:
1313 model = Document
1314 fields = (
1315 "id",
1316 "correspondent",
1317 "document_type",
1318 "storage_path",
1319 "title",
1320 "content",
1321 "tags",
1322 "created",
1323 "created_date",
1324 "modified",
1325 "added",
1326 "deleted_at",
1327 "archive_serial_number",
1328 "original_file_name",
1329 "archived_file_name",
1330 "duplicate_documents",
1331 "owner",
1332 "permissions",
1333 "user_can_change",
1334 "is_shared_by_requester",
1335 "set_permissions",
1336 "notes",
1337 "custom_fields",
1338 "remove_inbox_tags",
1339 "page_count",
1340 "mime_type",
1341 "root_document",
1342 "versions",
1343 )
1344 read_only_fields = ("deleted_at",)
1345 list_serializer_class = OwnedObjectListSerializer
1348class SearchResultListSerializer(serializers.ListSerializer[Document]):
1349 def to_representation(self, hits):
1350 document_ids = [hit["id"] for hit in hits]
1351 # Fetch all Document objects in the list in one SQL query.
1352 documents = self.child.fetch_documents(document_ids)
1353 self.child.context["documents"] = documents
1354 # Also check if they are shared with other users / groups.
1355 self.child.context["shared_object_pks"] = self.child.get_shared_object_pks(
1356 documents.values(),
1357 )
1359 return super().to_representation(hits)
1362class SearchResultSerializer(DocumentSerializer):
1363 @staticmethod
1364 def fetch_documents(ids):
1365 """
1366 Return a dict that maps given document IDs to Document objects.
1367 """
1368 return {
1369 document.id: document
1370 for document in Document.objects.select_related(
1371 "correspondent",
1372 "storage_path",
1373 "document_type",
1374 "owner",
1375 )
1376 .prefetch_related("tags", "custom_fields", "notes")
1377 .filter(id__in=ids)
1378 }
1380 def to_representation(self, hit):
1381 # Again we first check if the parent has already fetched the documents.
1382 documents = self.context.get("documents")
1383 # Otherwise we fetch this document.
1384 if documents is None: # pragma: no cover
1385 # In practice we only serialize **lists** of SearchHit dicts.
1386 # Keeping this check for completeness but marking it no cover for now.
1387 documents = self.fetch_documents([hit["id"]])
1388 document = documents[hit["id"]]
1390 highlights = hit.get("highlights", {})
1391 r = super().to_representation(document)
1392 r["__search_hit__"] = {
1393 "score": hit["score"],
1394 "highlights": highlights.get("content", ""),
1395 "note_highlights": highlights.get("notes") or None,
1396 "rank": hit["rank"],
1397 }
1399 return r
1401 class Meta(DocumentSerializer.Meta):
1402 list_serializer_class = SearchResultListSerializer
1405class SavedViewFilterRuleSerializer(serializers.ModelSerializer[SavedViewFilterRule]):
1406 class Meta:
1407 model = SavedViewFilterRule
1408 fields = ["rule_type", "value"]
1411class SavedViewSerializer(OwnedObjectSerializer):
1412 filter_rules = SavedViewFilterRuleSerializer(many=True)
1414 class Meta:
1415 model = SavedView
1416 fields = [
1417 "id",
1418 "name",
1419 "icon",
1420 "sort_field",
1421 "sort_reverse",
1422 "filter_rules",
1423 "page_size",
1424 "display_mode",
1425 "display_fields",
1426 "owner",
1427 "permissions",
1428 "user_can_change",
1429 "set_permissions",
1430 ]
1432 def _get_api_version(self) -> int:
1433 request = self.context.get("request")
1434 return int(
1435 request.version if request else settings.REST_FRAMEWORK["DEFAULT_VERSION"],
1436 )
1438 def _update_legacy_visibility_preferences(
1439 self,
1440 saved_view_id: int,
1441 *,
1442 show_on_dashboard: bool | None,
1443 show_in_sidebar: bool | None,
1444 ) -> UiSettings | None:
1445 if show_on_dashboard is None and show_in_sidebar is None: 1445 ↛ 1448line 1445 didn't jump to line 1448 because the condition on line 1445 was always true
1446 return None
1448 request = self.context.get("request")
1449 user = request.user if request else self.user
1450 if user is None:
1451 return None
1453 ui_settings, _ = UiSettings.objects.get_or_create(
1454 user=user,
1455 defaults={"settings": {}},
1456 )
1457 current_settings = (
1458 ui_settings.settings if isinstance(ui_settings.settings, dict) else {}
1459 )
1460 current_settings = dict(current_settings)
1462 saved_views_settings = current_settings.get("saved_views")
1463 if isinstance(saved_views_settings, dict):
1464 saved_views_settings = dict(saved_views_settings)
1465 else:
1466 saved_views_settings = {}
1468 dashboard_ids = {
1469 int(raw_id)
1470 for raw_id in saved_views_settings.get("dashboard_views_visible_ids", [])
1471 if str(raw_id).isdigit()
1472 }
1473 sidebar_ids = {
1474 int(raw_id)
1475 for raw_id in saved_views_settings.get("sidebar_views_visible_ids", [])
1476 if str(raw_id).isdigit()
1477 }
1479 if show_on_dashboard is not None:
1480 if show_on_dashboard:
1481 dashboard_ids.add(saved_view_id)
1482 else:
1483 dashboard_ids.discard(saved_view_id)
1484 if show_in_sidebar is not None:
1485 if show_in_sidebar:
1486 sidebar_ids.add(saved_view_id)
1487 else:
1488 sidebar_ids.discard(saved_view_id)
1490 saved_views_settings["dashboard_views_visible_ids"] = sorted(dashboard_ids)
1491 saved_views_settings["sidebar_views_visible_ids"] = sorted(sidebar_ids)
1492 current_settings["saved_views"] = saved_views_settings
1493 ui_settings.settings = current_settings
1494 ui_settings.save(update_fields=["settings"])
1495 return ui_settings
1497 def to_representation(self, instance):
1498 # TODO: remove this and related backwards compatibility code when API v9 is dropped
1499 ret = super().to_representation(instance)
1500 request = self.context.get("request")
1501 api_version = self._get_api_version()
1503 if api_version < 10: 1503 ↛ 1504line 1503 didn't jump to line 1504 because the condition on line 1503 was never true
1504 dashboard_ids = set()
1505 sidebar_ids = set()
1506 user = request.user if request else None
1507 if user is not None and hasattr(user, "ui_settings"):
1508 ui_settings = user.ui_settings.settings or None
1509 saved_views = None
1510 if isinstance(ui_settings, dict):
1511 saved_views = ui_settings.get("saved_views", {})
1512 if isinstance(saved_views, dict):
1513 dashboard_ids = set(
1514 saved_views.get("dashboard_views_visible_ids", []),
1515 )
1516 sidebar_ids = set(
1517 saved_views.get("sidebar_views_visible_ids", []),
1518 )
1519 ret["show_on_dashboard"] = instance.id in dashboard_ids
1520 ret["show_in_sidebar"] = instance.id in sidebar_ids
1522 return ret
1524 def to_internal_value(self, data):
1525 # TODO: remove this and related backwards compatibility code when API v9 is dropped
1526 api_version = self._get_api_version()
1527 if api_version >= 10: 1527 ↛ 1530line 1527 didn't jump to line 1530 because the condition on line 1527 was always true
1528 return super().to_internal_value(data)
1530 normalized_data = data.copy()
1531 legacy_visibility_fields = {}
1532 boolean_field = serializers.BooleanField()
1534 for field_name in ("show_on_dashboard", "show_in_sidebar"):
1535 if field_name in normalized_data:
1536 try:
1537 legacy_visibility_fields[field_name] = (
1538 boolean_field.to_internal_value(
1539 normalized_data.get(field_name),
1540 )
1541 )
1542 except serializers.ValidationError as exc:
1543 raise serializers.ValidationError({field_name: exc.detail})
1544 del normalized_data[field_name]
1546 ret = super().to_internal_value(normalized_data)
1547 ret.update(legacy_visibility_fields)
1548 return ret
1550 def validate(self, attrs):
1551 attrs = super().validate(attrs)
1552 if "display_fields" in attrs and attrs["display_fields"] is not None:
1553 for field in attrs["display_fields"]:
1554 if ( 1554 ↛ anywhereline 1554 didn't jump anywhere: it always raised an exception.
1555 SavedView.DisplayFields.CUSTOM_FIELD[:-2] in field
1556 ): # i.e. check for 'custom_field_' prefix
1557 field_id = int(re.search(r"\d+", field)[0])
1558 if not CustomField.objects.filter(id=field_id).exists():
1559 raise serializers.ValidationError(
1560 f"Invalid field: {field}",
1561 )
1562 elif field not in SavedView.DisplayFields.values:
1563 raise serializers.ValidationError(
1564 f"Invalid field: {field}",
1565 )
1566 return attrs
1568 def update(self, instance, validated_data):
1569 request = self.context.get("request")
1570 show_on_dashboard = validated_data.pop("show_on_dashboard", None)
1571 show_in_sidebar = validated_data.pop("show_in_sidebar", None)
1572 if "filter_rules" in validated_data:
1573 rules_data = validated_data.pop("filter_rules")
1574 else:
1575 rules_data = None
1576 if "user" in validated_data: 1576 ↛ 1578line 1576 didn't jump to line 1578 because the condition on line 1576 was never true
1577 # backwards compatibility
1578 validated_data["owner"] = validated_data.pop("user")
1579 if ( 1579 ↛ 1587line 1579 didn't jump to line 1587 because the condition on line 1579 was never true
1580 "display_fields" in validated_data
1581 and isinstance(
1582 validated_data["display_fields"],
1583 list,
1584 )
1585 and len(validated_data["display_fields"]) == 0
1586 ):
1587 validated_data["display_fields"] = None
1588 instance = super().update(instance, validated_data)
1589 if rules_data is not None:
1590 SavedViewFilterRule.objects.filter(saved_view=instance).delete()
1591 for rule_data in rules_data:
1592 SavedViewFilterRule.objects.create(saved_view=instance, **rule_data)
1593 ui_settings = self._update_legacy_visibility_preferences(
1594 instance.id,
1595 show_on_dashboard=show_on_dashboard,
1596 show_in_sidebar=show_in_sidebar,
1597 )
1598 if request is not None and ui_settings is not None: 1598 ↛ 1599line 1598 didn't jump to line 1599 because the condition on line 1598 was never true
1599 request.user.ui_settings = ui_settings
1600 return instance
1602 def create(self, validated_data):
1603 request = self.context.get("request")
1604 show_on_dashboard = validated_data.pop("show_on_dashboard", None)
1605 show_in_sidebar = validated_data.pop("show_in_sidebar", None)
1606 rules_data = validated_data.pop("filter_rules")
1607 if "user" in validated_data: 1607 ↛ 1609line 1607 didn't jump to line 1609 because the condition on line 1607 was never true
1608 # backwards compatibility
1609 validated_data["owner"] = validated_data.pop("user")
1610 saved_view = super().create(validated_data)
1611 for rule_data in rules_data:
1612 SavedViewFilterRule.objects.create(saved_view=saved_view, **rule_data)
1613 ui_settings = self._update_legacy_visibility_preferences(
1614 saved_view.id,
1615 show_on_dashboard=show_on_dashboard,
1616 show_in_sidebar=show_in_sidebar,
1617 )
1618 if request is not None and ui_settings is not None: 1618 ↛ 1619line 1618 didn't jump to line 1619 because the condition on line 1618 was never true
1619 request.user.ui_settings = ui_settings
1620 return saved_view
1623class DocumentListSerializer(serializers.Serializer[dict[str, list[int]]]):
1624 documents = serializers.ListField(
1625 required=True,
1626 label="Documents",
1627 write_only=True,
1628 child=serializers.IntegerField(),
1629 )
1631 def _validate_document_id_list(self, documents, name="documents") -> None:
1632 if not isinstance(documents, list): 1632 ↛ 1633line 1632 didn't jump to line 1633 because the condition on line 1632 was never true
1633 raise serializers.ValidationError(f"{name} must be a list")
1634 if not all(isinstance(i, int) for i in documents): 1634 ↛ 1635line 1634 didn't jump to line 1635 because the condition on line 1634 was never true
1635 raise serializers.ValidationError(f"{name} must be a list of integers")
1636 count = Document.objects.filter(id__in=documents).count()
1637 if not count == len(documents):
1638 raise serializers.ValidationError(
1639 f"Some documents in {name} don't exist or were specified twice.",
1640 )
1642 def validate_documents(self, documents):
1643 self._validate_document_id_list(documents)
1644 return documents
1647class DocumentSelectionSerializer(DocumentListSerializer):
1648 documents = serializers.ListField(
1649 required=False,
1650 label="Documents",
1651 write_only=True,
1652 child=serializers.IntegerField(),
1653 )
1655 all = serializers.BooleanField(
1656 default=False,
1657 required=False,
1658 write_only=True,
1659 )
1661 filters = serializers.DictField(
1662 required=False,
1663 allow_empty=True,
1664 write_only=True,
1665 )
1667 excluded_documents = serializers.ListField(
1668 required=False,
1669 default=list,
1670 write_only=True,
1671 child=serializers.IntegerField(),
1672 )
1674 def validate(self, attrs):
1675 if attrs.get("all", False):
1676 attrs.setdefault("documents", [])
1677 return attrs
1679 if attrs["excluded_documents"]:
1680 raise serializers.ValidationError(
1681 "excluded_documents is only supported when all is true.",
1682 )
1684 if "documents" not in attrs:
1685 raise serializers.ValidationError(
1686 "documents is required unless all is true.",
1687 )
1689 documents = attrs["documents"]
1690 self._validate_document_id_list(documents)
1691 return attrs
1694class SourceModeValidationMixin:
1695 def validate_source_mode(self, source_mode: str) -> str:
1696 if source_mode not in bulk_edit.SourceModeChoices.__dict__.values():
1697 raise serializers.ValidationError("Invalid source_mode")
1698 return source_mode
1701def _validate_rotation_degrees(degrees: int, field: str = "degrees") -> int:
1702 # QPDF refuses any other angle, which would otherwise fail inside the task
1703 if degrees % 90 != 0:
1704 raise serializers.ValidationError(f"{field} must be a multiple of 90")
1705 return degrees
1708class RotateDocumentsSerializer(DocumentSelectionSerializer, SourceModeValidationMixin):
1709 degrees = serializers.IntegerField(required=True)
1710 source_mode = serializers.CharField(
1711 required=False,
1712 default=bulk_edit.SourceModeChoices.LATEST_VERSION,
1713 )
1714 from_webui = serializers.BooleanField(required=False, default=False)
1716 def validate_degrees(self, value: int) -> int:
1717 return _validate_rotation_degrees(value)
1720class MergeDocumentsSerializer(DocumentListSerializer, SourceModeValidationMixin):
1721 metadata_document_id = serializers.IntegerField(
1722 required=False,
1723 allow_null=True,
1724 )
1725 delete_originals = serializers.BooleanField(required=False, default=False)
1726 archive_fallback = serializers.BooleanField(required=False, default=False)
1727 source_mode = serializers.CharField(
1728 required=False,
1729 default=bulk_edit.SourceModeChoices.LATEST_VERSION,
1730 )
1731 from_webui = serializers.BooleanField(required=False, default=False)
1734class MergeDocumentsAsVersionsSerializer(DocumentListSerializer):
1735 root_document_id = serializers.IntegerField(required=True)
1736 version_label = serializers.CharField(
1737 required=False,
1738 allow_blank=True,
1739 allow_null=True,
1740 max_length=64,
1741 )
1743 def validate_version_label(self, value):
1744 if value is None:
1745 return None
1746 normalized = value.strip()
1747 return normalized or None
1749 def validate(self, attrs):
1750 documents = attrs["documents"]
1751 if len(documents) < 2: 1751 ↛ 1755line 1751 didn't jump to line 1755 because the condition on line 1751 was always true
1752 raise serializers.ValidationError(
1753 "At least two documents are required.",
1754 )
1755 if attrs.get("version_label") is not None and len(documents) != 2:
1756 raise serializers.ValidationError(
1757 "version_label can only be used when merging one source document.",
1758 )
1759 if attrs["root_document_id"] not in documents:
1760 raise serializers.ValidationError(
1761 "root_document_id must be one of the selected documents.",
1762 )
1764 selected_documents = Document.objects.filter(id__in=documents)
1765 if selected_documents.filter(root_document__isnull=False).exists():
1766 raise serializers.ValidationError(
1767 "Only top-level documents can be merged as versions.",
1768 )
1770 source_document_ids = set(documents) - {attrs["root_document_id"]}
1771 if Document.global_objects.filter(
1772 root_document_id__in=source_document_ids,
1773 ).exists():
1774 raise serializers.ValidationError(
1775 "Documents with existing versions cannot be merged into another document.",
1776 )
1777 return attrs
1780class PdfEditOperationSerializer(serializers.Serializer[dict[str, int]]):
1781 page = serializers.IntegerField(min_value=1)
1782 rotate = serializers.IntegerField(required=False)
1783 doc = serializers.IntegerField(required=False, min_value=0)
1785 def validate_rotate(self, value: int) -> int:
1786 return _validate_rotation_degrees(value, field="rotate")
1789class EditPdfDocumentsSerializer(DocumentListSerializer, SourceModeValidationMixin):
1790 operations = serializers.ListField(
1791 child=PdfEditOperationSerializer(),
1792 required=True,
1793 allow_empty=False,
1794 )
1795 delete_original = serializers.BooleanField(required=False, default=False)
1796 update_document = serializers.BooleanField(required=False, default=False)
1797 include_metadata = serializers.BooleanField(required=False, default=True)
1798 source_mode = serializers.CharField(
1799 required=False,
1800 default=bulk_edit.SourceModeChoices.LATEST_VERSION,
1801 )
1802 from_webui = serializers.BooleanField(required=False, default=False)
1804 def validate(self, attrs):
1805 documents = attrs["documents"]
1806 if len(documents) > 1: 1806 ↛ 1807line 1806 didn't jump to line 1807 because the condition on line 1806 was never true
1807 raise serializers.ValidationError(
1808 "Edit PDF method only supports one document",
1809 )
1811 operations = attrs["operations"]
1813 if any(op.get("doc", 0) >= len(operations) for op in operations):
1814 raise serializers.ValidationError("doc index is out of bounds")
1816 if attrs["update_document"]: 1816 ↛ 1817line 1816 didn't jump to line 1817 because the condition on line 1816 was never true
1817 max_idx = max(op.get("doc", 0) for op in operations)
1818 if max_idx > 0:
1819 raise serializers.ValidationError(
1820 "update_document only allowed with a single output document",
1821 )
1823 doc = Document.objects.get(id=documents[0])
1824 if doc.page_count:
1825 for op in operations:
1826 if op["page"] > doc.page_count:
1827 raise serializers.ValidationError(
1828 f"Page {op['page']} is out of bounds for document with {doc.page_count} pages.",
1829 )
1830 return attrs
1833class RemovePasswordDocumentsSerializer(
1834 DocumentListSerializer,
1835 SourceModeValidationMixin,
1836):
1837 password = serializers.CharField(required=True)
1838 update_document = serializers.BooleanField(required=False, default=False)
1839 delete_original = serializers.BooleanField(required=False, default=False)
1840 include_metadata = serializers.BooleanField(required=False, default=True)
1841 source_mode = serializers.CharField(
1842 required=False,
1843 default=bulk_edit.SourceModeChoices.LATEST_VERSION,
1844 )
1845 from_webui = serializers.BooleanField(required=False, default=False)
1848class DeleteDocumentsSerializer(DocumentSelectionSerializer):
1849 pass
1852class ReprocessDocumentsSerializer(DocumentSelectionSerializer):
1853 remote_ocr = serializers.BooleanField(required=False, default=False)
1856class BulkEditSerializer(
1857 SerializerWithPerms,
1858 DocumentSelectionSerializer,
1859 SetPermissionsMixin,
1860 SourceModeValidationMixin,
1861):
1862 # TODO: remove this and related backwards compatibility code when API v9 is dropped
1863 # split, delete_pages can be removed entirely
1864 MOVED_DOCUMENT_ACTION_ENDPOINTS = {
1865 "delete": "/api/documents/delete/",
1866 "reprocess": "/api/documents/reprocess/",
1867 "rotate": "/api/documents/rotate/",
1868 "merge": "/api/documents/merge/",
1869 "edit_pdf": "/api/documents/edit_pdf/",
1870 "remove_password": "/api/documents/remove_password/",
1871 "split": "/api/documents/edit_pdf/",
1872 "delete_pages": "/api/documents/edit_pdf/",
1873 }
1874 LEGACY_DOCUMENT_ACTION_METHODS = tuple(MOVED_DOCUMENT_ACTION_ENDPOINTS.keys())
1876 method = serializers.ChoiceField(
1877 choices=[
1878 "set_correspondent",
1879 "set_document_type",
1880 "set_storage_path",
1881 "add_tag",
1882 "remove_tag",
1883 "modify_tags",
1884 "modify_custom_fields",
1885 "set_permissions",
1886 *LEGACY_DOCUMENT_ACTION_METHODS,
1887 ],
1888 label="Method",
1889 write_only=True,
1890 )
1892 parameters = serializers.DictField(allow_empty=True, default={}, write_only=True)
1893 from_webui = serializers.BooleanField(required=False, default=False)
1895 def _validate_tag_id_list(self, tags, name="tags") -> None:
1896 if not isinstance(tags, list):
1897 raise serializers.ValidationError(f"{name} must be a list")
1898 if not all(isinstance(i, int) for i in tags):
1899 raise serializers.ValidationError(f"{name} must be a list of integers")
1900 count = Tag.objects.filter(id__in=tags).count()
1901 if not count == len(tags):
1902 raise serializers.ValidationError(
1903 f"Some tags in {name} don't exist or were specified twice.",
1904 )
1906 def _validate_custom_field_id_list_or_dict(
1907 self,
1908 custom_fields,
1909 name="custom_fields",
1910 ) -> None:
1911 ids = custom_fields
1912 if isinstance(custom_fields, dict):
1913 try:
1914 ids = [int(i[0]) for i in custom_fields.items()]
1915 except Exception as e:
1916 logger.exception(f"Error validating custom fields: {e}")
1917 raise serializers.ValidationError(
1918 f"{name} must be a list of integers or a dict of id:value pairs, see the log for details",
1919 )
1920 elif not isinstance(custom_fields, list) or not all(
1921 isinstance(i, int) for i in ids
1922 ):
1923 raise serializers.ValidationError(
1924 f"{name} must be a list of integers or a dict of id:value pairs",
1925 )
1926 count = CustomField.objects.filter(id__in=ids).count()
1927 if not count == len(ids):
1928 raise serializers.ValidationError(
1929 f"Some custom fields in {name} don't exist or were specified twice.",
1930 )
1932 def _validate_custom_field_values(self, custom_fields, name):
1933 if not isinstance(custom_fields, dict):
1934 return custom_fields
1936 validated = {}
1937 errors = {}
1938 for raw_field_id, value in custom_fields.items():
1939 field_id = int(raw_field_id)
1940 validator = CustomFieldInstanceSerializer(
1941 data={"field": field_id, "value": value},
1942 context=self.context,
1943 )
1944 if validator.is_valid():
1945 validated[field_id] = validator.validated_data["value"]
1946 else:
1947 errors[str(field_id)] = validator.errors
1949 if errors:
1950 raise serializers.ValidationError({name: errors})
1952 return validated
1954 def validate_method(self, method):
1955 if method == "set_correspondent":
1956 return bulk_edit.set_correspondent
1957 elif method == "set_document_type":
1958 return bulk_edit.set_document_type
1959 elif method == "set_storage_path":
1960 return bulk_edit.set_storage_path
1961 elif method == "add_tag":
1962 return bulk_edit.add_tag
1963 elif method == "remove_tag":
1964 return bulk_edit.remove_tag
1965 elif method == "modify_tags":
1966 return bulk_edit.modify_tags
1967 elif method == "modify_custom_fields":
1968 return bulk_edit.modify_custom_fields
1969 elif method == "delete":
1970 return bulk_edit.delete
1971 elif method == "redo_ocr" or method == "reprocess":
1972 return bulk_edit.reprocess
1973 elif method == "set_permissions":
1974 return bulk_edit.set_permissions
1975 elif method == "rotate":
1976 return bulk_edit.rotate
1977 elif method == "merge":
1978 return bulk_edit.merge
1979 elif method == "split":
1980 return bulk_edit.split
1981 elif method == "delete_pages":
1982 return bulk_edit.delete_pages
1983 elif method == "edit_pdf":
1984 return bulk_edit.edit_pdf
1985 elif method == "remove_password": 1985 ↛ 1988line 1985 didn't jump to line 1988 because the condition on line 1985 was always true
1986 return bulk_edit.remove_password
1987 else:
1988 raise serializers.ValidationError("Unsupported method.")
1990 def _validate_parameters_tags(self, parameters) -> None:
1991 if "tag" in parameters: 1991 ↛ 1992line 1991 didn't jump to line 1992 because the condition on line 1991 was never true
1992 tag_id = parameters["tag"]
1993 try:
1994 Tag.objects.get(id=tag_id)
1995 except Tag.DoesNotExist:
1996 raise serializers.ValidationError("Tag does not exist")
1997 else:
1998 raise serializers.ValidationError("tag not specified")
2000 def _validate_parameters_document_type(self, parameters) -> None:
2001 if "document_type" in parameters: 2001 ↛ 2002line 2001 didn't jump to line 2002 because the condition on line 2001 was never true
2002 document_type_id = parameters["document_type"]
2003 if document_type_id is None:
2004 # None is ok
2005 return
2006 try:
2007 DocumentType.objects.get(id=document_type_id)
2008 except DocumentType.DoesNotExist:
2009 raise serializers.ValidationError("Document type does not exist")
2010 else:
2011 raise serializers.ValidationError("document_type not specified")
2013 def _validate_parameters_correspondent(self, parameters) -> None:
2014 if "correspondent" in parameters: 2014 ↛ 2015line 2014 didn't jump to line 2015 because the condition on line 2014 was never true
2015 correspondent_id = parameters["correspondent"]
2016 if correspondent_id is None:
2017 return
2018 try:
2019 Correspondent.objects.get(id=correspondent_id)
2020 except Correspondent.DoesNotExist:
2021 raise serializers.ValidationError("Correspondent does not exist")
2022 else:
2023 raise serializers.ValidationError("correspondent not specified")
2025 def _validate_storage_path(self, parameters) -> None:
2026 if "storage_path" in parameters: 2026 ↛ 2027line 2026 didn't jump to line 2027 because the condition on line 2026 was never true
2027 storage_path_id = parameters["storage_path"]
2028 if storage_path_id is None:
2029 return
2030 try:
2031 StoragePath.objects.get(id=storage_path_id)
2032 except StoragePath.DoesNotExist:
2033 raise serializers.ValidationError(
2034 "Storage path does not exist",
2035 )
2036 else:
2037 raise serializers.ValidationError("storage path not specified")
2039 def _validate_parameters_modify_tags(self, parameters) -> None:
2040 if "add_tags" in parameters: 2040 ↛ 2041line 2040 didn't jump to line 2041 because the condition on line 2040 was never true
2041 self._validate_tag_id_list(parameters["add_tags"], "add_tags")
2042 else:
2043 raise serializers.ValidationError("add_tags not specified")
2045 if "remove_tags" in parameters:
2046 self._validate_tag_id_list(parameters["remove_tags"], "remove_tags")
2047 else:
2048 raise serializers.ValidationError("remove_tags not specified")
2050 def _validate_parameters_modify_custom_fields(self, parameters) -> None:
2051 if "add_custom_fields" in parameters: 2051 ↛ 2052line 2051 didn't jump to line 2052 because the condition on line 2051 was never true
2052 self._validate_custom_field_id_list_or_dict(
2053 parameters["add_custom_fields"],
2054 "add_custom_fields",
2055 )
2056 parameters["add_custom_fields"] = self._validate_custom_field_values(
2057 parameters["add_custom_fields"],
2058 "add_custom_fields",
2059 )
2060 else:
2061 raise serializers.ValidationError("add_custom_fields not specified")
2063 if "remove_custom_fields" in parameters:
2064 self._validate_custom_field_id_list_or_dict(
2065 parameters["remove_custom_fields"],
2066 "remove_custom_fields",
2067 )
2068 else:
2069 raise serializers.ValidationError("remove_custom_fields not specified")
2071 def _validate_owner(self, owner) -> User:
2072 owner_field = serializers.PrimaryKeyRelatedField(queryset=User.objects.all())
2073 try:
2074 return owner_field.run_validation(owner)
2075 except serializers.ValidationError as e:
2076 raise serializers.ValidationError(
2077 "Specified owner cannot be found",
2078 ) from e
2080 def _validate_parameters_set_permissions(self, parameters) -> None:
2081 if "set_permissions" not in parameters: 2081 ↛ 2083line 2081 didn't jump to line 2083 because the condition on line 2081 was always true
2082 raise serializers.ValidationError("set_permissions not specified")
2083 set_permissions = parameters["set_permissions"]
2084 if set_permissions is not None:
2085 set_permissions = SetPermissionsSerializer().run_validation(
2086 set_permissions,
2087 )
2088 parameters["set_permissions"] = self.validate_set_permissions(
2089 set_permissions,
2090 )
2091 if "owner" in parameters and parameters["owner"] is not None:
2092 parameters["owner"] = self._validate_owner(parameters["owner"]).pk
2093 if "merge" not in parameters:
2094 parameters["merge"] = False
2096 def _validate_parameters_rotate(self, parameters) -> None:
2097 if "degrees" not in parameters: 2097 ↛ 2099line 2097 didn't jump to line 2099 because the condition on line 2097 was always true
2098 raise serializers.ValidationError("invalid rotation degrees")
2099 try:
2100 degrees = serializers.IntegerField().run_validation(parameters["degrees"])
2101 except serializers.ValidationError as e:
2102 raise serializers.ValidationError("invalid rotation degrees") from e
2103 parameters["degrees"] = _validate_rotation_degrees(degrees)
2105 def _validate_source_mode(self, parameters) -> None:
2106 source_mode = parameters.get(
2107 "source_mode",
2108 bulk_edit.SourceModeChoices.LATEST_VERSION,
2109 )
2110 parameters["source_mode"] = self.validate_source_mode(source_mode)
2112 def _validate_parameters_split(self, parameters, document_id) -> None:
2113 if "pages" not in parameters:
2114 raise serializers.ValidationError("pages not specified")
2115 if not isinstance(parameters["pages"], str):
2116 raise serializers.ValidationError("invalid pages specified")
2117 page_count = Document.objects.get(id=document_id).page_count
2118 if not page_count:
2119 raise serializers.ValidationError("document page count is unknown")
2120 pages = []
2121 for group in parameters["pages"].split(","):
2122 start, is_range, end = group.partition("-")
2123 try:
2124 first = int(start)
2125 last = int(end) if is_range else first
2126 except ValueError as e:
2127 raise serializers.ValidationError("invalid pages specified") from e
2128 # Bound the range before building it, a huge one would exhaust memory
2129 if not 1 <= first <= last <= page_count:
2130 raise serializers.ValidationError("invalid pages specified")
2131 pages.append(list(range(first, last + 1)))
2132 parameters["pages"] = pages
2134 if "delete_originals" in parameters:
2135 if not isinstance(parameters["delete_originals"], bool):
2136 raise serializers.ValidationError("delete_originals must be a boolean")
2137 else:
2138 parameters["delete_originals"] = False
2140 def _validate_parameters_delete_pages(self, parameters) -> None:
2141 if "pages" not in parameters: 2141 ↛ 2143line 2141 didn't jump to line 2143 because the condition on line 2141 was always true
2142 raise serializers.ValidationError("pages not specified")
2143 if not isinstance(parameters["pages"], list):
2144 raise serializers.ValidationError("pages must be a list")
2145 if not all(isinstance(i, int) for i in parameters["pages"]):
2146 raise serializers.ValidationError("pages must be a list of integers")
2148 def _validate_parameters_merge(self, parameters) -> None:
2149 if "delete_originals" in parameters: 2149 ↛ 2150line 2149 didn't jump to line 2150 because the condition on line 2149 was never true
2150 if not isinstance(parameters["delete_originals"], bool):
2151 raise serializers.ValidationError("delete_originals must be a boolean")
2152 else:
2153 parameters["delete_originals"] = False
2154 if "archive_fallback" in parameters: 2154 ↛ 2155line 2154 didn't jump to line 2155 because the condition on line 2154 was never true
2155 if not isinstance(parameters["archive_fallback"], bool):
2156 raise serializers.ValidationError("archive_fallback must be a boolean")
2157 else:
2158 parameters["archive_fallback"] = False
2160 def _validate_parameters_edit_pdf(self, parameters, document_id) -> None:
2161 if "operations" not in parameters:
2162 raise serializers.ValidationError("operations not specified")
2163 operations_field = serializers.ListField(
2164 child=PdfEditOperationSerializer(),
2165 allow_empty=False,
2166 )
2167 try:
2168 operations = operations_field.run_validation(parameters["operations"])
2169 except serializers.ValidationError as e:
2170 # Key the errors under "operations" so they match what the
2171 # dedicated edit_pdf endpoint returns
2172 raise serializers.ValidationError({"operations": e.detail}) from e
2173 parameters["operations"] = operations
2175 if "update_document" in parameters:
2176 if not isinstance(parameters["update_document"], bool):
2177 raise serializers.ValidationError("update_document must be a boolean")
2178 else:
2179 parameters["update_document"] = False
2180 if "include_metadata" in parameters:
2181 if not isinstance(parameters["include_metadata"], bool):
2182 raise serializers.ValidationError("include_metadata must be a boolean")
2183 else:
2184 parameters["include_metadata"] = True
2186 if any(op.get("doc", 0) >= len(operations) for op in operations):
2187 raise serializers.ValidationError("doc index is out of bounds")
2189 if parameters["update_document"]:
2190 max_idx = max(op.get("doc", 0) for op in operations)
2191 if max_idx > 0:
2192 raise serializers.ValidationError(
2193 "update_document only allowed with a single output document",
2194 )
2196 doc = Document.objects.get(id=document_id)
2197 # doc existence is already validated
2198 if doc.page_count:
2199 for op in operations:
2200 if op["page"] > doc.page_count:
2201 raise serializers.ValidationError(
2202 f"Page {op['page']} is out of bounds for document with {doc.page_count} pages.",
2203 )
2205 def _validate_parameters_reprocess(self, parameters) -> None:
2206 if "remote_ocr" in parameters: 2206 ↛ 2207line 2206 didn't jump to line 2207 because the condition on line 2206 was never true
2207 if not isinstance(parameters["remote_ocr"], bool):
2208 raise serializers.ValidationError("remote_ocr must be a boolean")
2209 else:
2210 parameters["remote_ocr"] = False
2212 def validate_parameters_remove_password(self, parameters):
2213 if "password" not in parameters: 2213 ↛ 2215line 2213 didn't jump to line 2215 because the condition on line 2213 was always true
2214 raise serializers.ValidationError("password not specified")
2215 if not isinstance(parameters["password"], str):
2216 raise serializers.ValidationError("password must be a string")
2218 def validate(self, attrs):
2219 attrs = super().validate(attrs)
2221 if attrs.get("all", False) and attrs["method"] in [
2222 bulk_edit.merge,
2223 bulk_edit.split,
2224 bulk_edit.delete_pages,
2225 bulk_edit.edit_pdf,
2226 bulk_edit.remove_password,
2227 ]:
2228 raise serializers.ValidationError(
2229 "This method does not support all=true.",
2230 )
2232 method = attrs["method"]
2233 parameters = attrs["parameters"]
2235 if "source_mode" in parameters: 2235 ↛ 2236line 2235 didn't jump to line 2236 because the condition on line 2235 was never true
2236 self._validate_source_mode(parameters)
2238 if method == bulk_edit.set_correspondent:
2239 self._validate_parameters_correspondent(parameters)
2240 elif method == bulk_edit.set_document_type:
2241 self._validate_parameters_document_type(parameters)
2242 elif method == bulk_edit.add_tag or method == bulk_edit.remove_tag:
2243 self._validate_parameters_tags(parameters)
2244 elif method == bulk_edit.modify_tags:
2245 self._validate_parameters_modify_tags(parameters)
2246 elif method == bulk_edit.set_storage_path:
2247 self._validate_storage_path(parameters)
2248 elif method == bulk_edit.modify_custom_fields:
2249 self._validate_parameters_modify_custom_fields(parameters)
2250 elif method == bulk_edit.set_permissions:
2251 self._validate_parameters_set_permissions(parameters)
2252 elif method == bulk_edit.rotate:
2253 self._validate_parameters_rotate(parameters)
2254 elif method == bulk_edit.split:
2255 if len(attrs["documents"]) > 1: 2255 ↛ 2256line 2255 didn't jump to line 2256 because the condition on line 2255 was never true
2256 raise serializers.ValidationError(
2257 "Split method only supports one document",
2258 )
2259 self._validate_parameters_split(parameters, attrs["documents"][0])
2260 elif method == bulk_edit.delete_pages:
2261 if len(attrs["documents"]) > 1: 2261 ↛ 2262line 2261 didn't jump to line 2262 because the condition on line 2261 was never true
2262 raise serializers.ValidationError(
2263 "Delete pages method only supports one document",
2264 )
2265 self._validate_parameters_delete_pages(parameters)
2266 elif method == bulk_edit.merge:
2267 self._validate_parameters_merge(parameters)
2268 elif method == bulk_edit.edit_pdf:
2269 if len(attrs["documents"]) > 1: 2269 ↛ 2270line 2269 didn't jump to line 2270 because the condition on line 2269 was never true
2270 raise serializers.ValidationError(
2271 "Edit PDF method only supports one document",
2272 )
2273 self._validate_parameters_edit_pdf(parameters, attrs["documents"][0])
2274 elif method == bulk_edit.remove_password:
2275 self.validate_parameters_remove_password(parameters)
2276 elif method == bulk_edit.reprocess:
2277 self._validate_parameters_reprocess(parameters)
2279 return attrs
2282class PostDocumentSerializer(serializers.Serializer[dict[str, Any]]):
2283 created = serializers.DateTimeField(
2284 label="Created",
2285 allow_null=True,
2286 write_only=True,
2287 required=False,
2288 )
2290 document = serializers.FileField(
2291 label="Document",
2292 write_only=True,
2293 )
2295 title = serializers.CharField(
2296 label="Title",
2297 write_only=True,
2298 required=False,
2299 )
2301 correspondent = serializers.PrimaryKeyRelatedField(
2302 queryset=Correspondent.objects.all(),
2303 label="Correspondent",
2304 allow_null=True,
2305 write_only=True,
2306 required=False,
2307 )
2309 document_type = serializers.PrimaryKeyRelatedField(
2310 queryset=DocumentType.objects.all(),
2311 label="Document type",
2312 allow_null=True,
2313 write_only=True,
2314 required=False,
2315 )
2317 storage_path = serializers.PrimaryKeyRelatedField(
2318 queryset=StoragePath.objects.all(),
2319 label="Storage path",
2320 allow_null=True,
2321 write_only=True,
2322 required=False,
2323 )
2325 tags = serializers.PrimaryKeyRelatedField(
2326 many=True,
2327 queryset=Tag.objects.all(),
2328 label="Tags",
2329 write_only=True,
2330 required=False,
2331 )
2333 archive_serial_number = serializers.IntegerField(
2334 label="ASN",
2335 write_only=True,
2336 required=False,
2337 min_value=Document.ARCHIVE_SERIAL_NUMBER_MIN,
2338 max_value=Document.ARCHIVE_SERIAL_NUMBER_MAX,
2339 )
2341 # Accept either a list of custom field ids or a dict mapping id -> value
2342 custom_fields = serializers.JSONField(
2343 label="Custom fields",
2344 write_only=True,
2345 required=False,
2346 )
2348 from_webui = serializers.BooleanField(
2349 label="Documents are from Paperless-ngx WebUI",
2350 write_only=True,
2351 required=False,
2352 )
2354 def validate_document(self, document):
2355 document_data = document.file.read()
2356 mime_type = magic.from_buffer(document_data, mime=True)
2358 if not is_mime_type_supported(mime_type):
2359 if ( 2359 ↛ 2366line 2359 didn't jump to line 2366 because the condition on line 2359 was never true
2360 mime_type in settings.CONSUMER_PDF_RECOVERABLE_MIME_TYPES
2361 and document.name.endswith(
2362 ".pdf",
2363 )
2364 ):
2365 # If the file is an invalid PDF, we can try to recover it later in the consumer
2366 mime_type = "application/pdf"
2367 else:
2368 raise serializers.ValidationError(
2369 _("File type %(type)s not supported") % {"type": mime_type},
2370 )
2372 return document.name, document_data
2374 def validate_correspondent(self, correspondent):
2375 if correspondent:
2376 return correspondent.id
2377 else:
2378 return None
2380 def validate_document_type(self, document_type):
2381 if document_type: 2381 ↛ 2384line 2381 didn't jump to line 2384 because the condition on line 2381 was always true
2382 return document_type.id
2383 else:
2384 return None
2386 def validate_storage_path(self, storage_path):
2387 if storage_path: 2387 ↛ 2390line 2387 didn't jump to line 2390 because the condition on line 2387 was always true
2388 return storage_path.id
2389 else:
2390 return None
2392 def validate_tags(self, tags):
2393 if tags: 2393 ↛ 2394line 2393 didn't jump to line 2394 because the condition on line 2393 was never true
2394 return [tag.id for tag in tags]
2395 else:
2396 return None
2398 def validate_custom_fields(self, custom_fields):
2399 if not custom_fields:
2400 return None
2402 # Normalize single values to a list
2403 if isinstance(custom_fields, int): 2403 ↛ 2405line 2403 didn't jump to line 2405 because the condition on line 2403 was always true
2404 custom_fields = [custom_fields]
2405 if isinstance(custom_fields, dict): 2405 ↛ 2406line 2405 didn't jump to line 2406 because the condition on line 2405 was never true
2406 custom_field_serializer = CustomFieldInstanceSerializer()
2407 normalized = {}
2408 for field_id, value in custom_fields.items():
2409 try:
2410 field_id_int = int(field_id)
2411 except (TypeError, ValueError):
2412 raise serializers.ValidationError(
2413 _("Custom field id must be an integer: %(id)s")
2414 % {"id": field_id},
2415 )
2416 try:
2417 field = CustomField.objects.get(id=field_id_int)
2418 except CustomField.DoesNotExist:
2419 raise serializers.ValidationError(
2420 _("Custom field with id %(id)s does not exist")
2421 % {"id": field_id_int},
2422 )
2423 custom_field_serializer.validate(
2424 {
2425 "field": field,
2426 "value": value,
2427 },
2428 )
2429 normalized[field_id_int] = value
2430 return normalized
2431 elif isinstance(custom_fields, list): 2431 ↛ 2445line 2431 didn't jump to line 2445 because the condition on line 2431 was always true
2432 try:
2433 ids = [int(i) for i in custom_fields]
2434 except (TypeError, ValueError):
2435 raise serializers.ValidationError(
2436 _(
2437 "Custom fields must be a list of integers or an object mapping ids to values.",
2438 ),
2439 )
2440 if CustomField.objects.filter(id__in=ids).count() != len(set(ids)):
2441 raise serializers.ValidationError(
2442 _("Some custom fields don't exist or were specified twice."),
2443 )
2444 return ids
2445 raise serializers.ValidationError(
2446 _(
2447 "Custom fields must be a list of integers or an object mapping ids to values.",
2448 ),
2449 )
2451 # custom_fields_w_values handled via validate_custom_fields
2453 def validate_created(self, created):
2454 # support datetime format for created for backwards compatibility
2455 if isinstance(created, datetime):
2456 return created.date()
2459class DocumentVersionSerializer(serializers.Serializer[dict[str, Any]]):
2460 document = serializers.FileField(
2461 label="Document",
2462 write_only=True,
2463 )
2464 version_label = serializers.CharField(
2465 label="Version label",
2466 required=False,
2467 allow_blank=True,
2468 allow_null=True,
2469 max_length=64,
2470 )
2472 validate_document = PostDocumentSerializer().validate_document
2475class DocumentVersionLabelSerializer(serializers.Serializer[dict[str, str | None]]):
2476 version_label = serializers.CharField(
2477 label="Version label",
2478 required=True,
2479 allow_blank=True,
2480 allow_null=True,
2481 max_length=64,
2482 )
2484 def validate_version_label(self, value):
2485 if value is None:
2486 return None
2487 normalized = value.strip()
2488 return normalized or None
2491class BulkDownloadSerializer(DocumentSelectionSerializer):
2492 content = serializers.ChoiceField(
2493 choices=["archive", "originals", "both"],
2494 default="archive",
2495 )
2497 compression = serializers.ChoiceField(
2498 choices=["none", "deflated", "bzip2", "lzma"],
2499 default="none",
2500 )
2502 follow_formatting = serializers.BooleanField(
2503 default=False,
2504 )
2506 def validate_compression(self, compression):
2507 import zipfile
2509 return {
2510 "none": zipfile.ZIP_STORED,
2511 "deflated": zipfile.ZIP_DEFLATED,
2512 "bzip2": zipfile.ZIP_BZIP2,
2513 "lzma": zipfile.ZIP_LZMA,
2514 }[compression]
2517class EmailSerializer(DocumentListSerializer):
2518 addresses = serializers.CharField(
2519 required=True,
2520 label="Email addresses",
2521 help_text="Comma-separated email addresses",
2522 )
2524 subject = serializers.CharField(
2525 required=True,
2526 label="Email subject",
2527 )
2529 message = serializers.CharField(
2530 required=True,
2531 label="Email message",
2532 )
2534 use_archive_version = serializers.BooleanField(
2535 default=True,
2536 label="Use archive version",
2537 help_text="Use archive version of documents if available",
2538 )
2540 def validate_addresses(self, addresses):
2541 address_list = [addr.strip() for addr in addresses.split(",")]
2542 if not address_list: 2542 ↛ 2543line 2542 didn't jump to line 2543 because the condition on line 2542 was never true
2543 raise serializers.ValidationError("At least one email address is required")
2545 email_validator = EmailValidator()
2546 try:
2547 for address in address_list: 2547 ↛ 2552line 2547 didn't jump to line 2552 because the loop on line 2547 didn't complete
2548 email_validator(address)
2549 except ValidationError:
2550 raise serializers.ValidationError(f"Invalid email address: {address}")
2552 return ",".join(address_list)
2554 def validate_documents(self, documents):
2555 super().validate_documents(documents)
2556 if not documents: 2556 ↛ 2559line 2556 didn't jump to line 2559 because the condition on line 2556 was always true
2557 raise serializers.ValidationError("At least one document is required")
2559 return documents
2562class StoragePathSerializer(MatchingModelSerializer, OwnedObjectSerializer):
2563 class Meta:
2564 model = StoragePath
2565 fields = (
2566 "id",
2567 "slug",
2568 "name",
2569 "path",
2570 "match",
2571 "matching_algorithm",
2572 "is_insensitive",
2573 "document_count",
2574 "owner",
2575 "permissions",
2576 "user_can_change",
2577 "set_permissions",
2578 )
2580 def validate_path(self, path: str):
2581 converted_path = convert_format_str_to_template_format(path)
2582 if converted_path != path: 2582 ↛ 2583line 2582 didn't jump to line 2583 because the condition on line 2582 was never true
2583 logger.warning(
2584 f"Storage path {path} is not using the new style format, consider updating",
2585 )
2586 result = validate_filepath_template_and_render(converted_path)
2588 if result is None: 2588 ↛ 2589line 2588 didn't jump to line 2589 because the condition on line 2588 was never true
2589 raise serializers.ValidationError(_("Invalid variable detected."))
2591 return converted_path
2593 def update(self, instance, validated_data):
2594 """
2595 When a storage path is updated, see if documents
2596 using it require a rename/move
2597 """
2598 doc_ids = [doc.id for doc in instance.documents.all()]
2599 if doc_ids: 2599 ↛ 2600line 2599 didn't jump to line 2600 because the condition on line 2599 was never true
2600 bulk_edit.bulk_update_documents.apply_async(
2601 kwargs={"document_ids": doc_ids},
2602 headers={"trigger_source": PaperlessTask.TriggerSource.SYSTEM},
2603 )
2605 return super().update(instance, validated_data)
2608class UiSettingsViewSerializer(serializers.ModelSerializer[UiSettings]):
2609 settings = serializers.DictField(required=False, allow_null=True)
2611 class Meta:
2612 model = UiSettings
2613 depth = 1
2614 fields = [
2615 "id",
2616 "settings",
2617 ]
2619 def validate_settings(self, settings):
2620 # we never save update checking backend setting
2621 if "update_checking" in settings: 2621 ↛ 2622line 2621 didn't jump to line 2622 because the condition on line 2621 was never true
2622 try:
2623 settings["update_checking"].pop("backend_setting")
2624 except KeyError:
2625 pass
2626 return settings
2628 def create(self, validated_data):
2629 ui_settings = UiSettings.objects.update_or_create(
2630 user=validated_data.get("user"),
2631 defaults={"settings": validated_data.get("settings", None)},
2632 )
2633 return ui_settings
2636class TaskSerializerV10(OwnedObjectSerializer):
2637 """Task serializer for API v10+ using new field names."""
2639 related_document_ids = serializers.ListField(
2640 child=serializers.IntegerField(),
2641 read_only=True,
2642 )
2643 task_type_display = serializers.CharField(
2644 source="get_task_type_display",
2645 read_only=True,
2646 )
2647 trigger_source_display = serializers.CharField(
2648 source="get_trigger_source_display",
2649 read_only=True,
2650 )
2651 status_display = serializers.CharField(
2652 source="get_status_display",
2653 read_only=True,
2654 )
2656 class Meta:
2657 model = PaperlessTask
2658 fields = (
2659 "id",
2660 "task_id",
2661 "task_type",
2662 "task_type_display",
2663 "trigger_source",
2664 "trigger_source_display",
2665 "status",
2666 "status_display",
2667 "date_created",
2668 "date_started",
2669 "date_done",
2670 "duration_seconds",
2671 "wait_time_seconds",
2672 "input_data",
2673 "result_data",
2674 "related_document_ids",
2675 "acknowledged",
2676 "owner",
2677 )
2678 read_only_fields = fields
2681class TaskSerializerV9(serializers.ModelSerializer[PaperlessTask]):
2682 """Task serializer for API v9 backwards compatibility.
2684 Maps old field names to the new model fields so existing clients continue
2685 to work unchanged.
2686 """
2688 # v9 field: task_name -> task_type (with value remapping for renamed tasks)
2689 task_name = serializers.SerializerMethodField()
2691 # v9 field: task_file_name -> input_data.filename
2692 task_file_name = serializers.SerializerMethodField()
2694 # v9 field: type -> trigger_source (mapped to old enum labels)
2695 type = serializers.SerializerMethodField()
2697 # v9 field: status -> uppercase Celery state strings
2698 status = serializers.SerializerMethodField()
2700 # v9 field: result -> derived from result_data
2701 result = serializers.SerializerMethodField()
2703 # v9 field: related_document -> first document ID from result_data
2704 related_document = serializers.SerializerMethodField()
2706 # v9 field: duplicate_documents -> list of duplicate IDs from result_data
2707 duplicate_documents = serializers.SerializerMethodField()
2709 class Meta:
2710 model = PaperlessTask
2711 fields = (
2712 "id",
2713 "task_id",
2714 "task_name",
2715 "task_file_name",
2716 "type",
2717 "status",
2718 "date_created",
2719 "date_done",
2720 "result",
2721 "acknowledged",
2722 "related_document",
2723 "duplicate_documents",
2724 "owner",
2725 )
2726 read_only_fields = fields
2728 _TASK_TYPE_TO_V9_NAME = {
2729 PaperlessTask.TaskType.SANITY_CHECK: "check_sanity",
2730 PaperlessTask.TaskType.LLM_INDEX: "llmindex_update",
2731 }
2733 def get_result(self, obj: PaperlessTask) -> str | None:
2734 """Reconstruct a human-readable result string from result_data for v9 clients."""
2735 if not obj.result_data:
2736 return None
2737 if doc_id := obj.result_data.get("document_id"):
2738 return f"Success. New document id {doc_id} created"
2739 if reason := obj.result_data.get("reason"):
2740 return reason
2741 if dup_id := obj.result_data.get("duplicate_of"):
2742 return f"Not consuming: It is a duplicate of document #{dup_id}"
2743 if error := obj.result_data.get("error_message"):
2744 return error
2745 return None
2747 def get_task_name(self, obj: PaperlessTask) -> str:
2748 return self._TASK_TYPE_TO_V9_NAME.get(obj.task_type, obj.task_type)
2750 def get_task_file_name(self, obj: PaperlessTask) -> str | None:
2751 if not obj.input_data:
2752 return None
2753 return obj.input_data.get("filename")
2755 _STATUS_TO_V9 = {
2756 PaperlessTask.Status.PENDING: "PENDING",
2757 PaperlessTask.Status.STARTED: "STARTED",
2758 PaperlessTask.Status.SUCCESS: "SUCCESS",
2759 PaperlessTask.Status.FAILURE: "FAILURE",
2760 PaperlessTask.Status.REVOKED: "REVOKED",
2761 }
2763 def get_status(self, obj: PaperlessTask) -> str:
2764 return self._STATUS_TO_V9.get(obj.status, obj.status.upper())
2766 _TRIGGER_SOURCE_TO_V9_TYPE = {
2767 PaperlessTask.TriggerSource.SCHEDULED: "scheduled_task",
2768 PaperlessTask.TriggerSource.SYSTEM: "auto_task",
2769 # Email and folder-consumer documents are system-initiated, not manually triggered
2770 PaperlessTask.TriggerSource.EMAIL_CONSUME: "auto_task",
2771 PaperlessTask.TriggerSource.FOLDER_CONSUME: "auto_task",
2772 }
2774 def get_type(self, obj: PaperlessTask) -> str:
2775 return self._TRIGGER_SOURCE_TO_V9_TYPE.get(obj.trigger_source, "manual_task")
2777 def get_related_document(self, obj: PaperlessTask) -> int | None:
2778 ids = obj.related_document_ids
2779 return ids[0] if ids else None
2781 def get_duplicate_documents(
2782 self,
2783 obj: PaperlessTask,
2784 ) -> list[dict[str, Any]]:
2785 if not obj.result_data:
2786 return []
2787 dup_of = obj.result_data.get("duplicate_of")
2788 if dup_of is None:
2789 return []
2790 request = self.context.get("request")
2791 if request is None:
2792 return []
2793 user = request.user
2794 qs = Document.global_objects.filter(pk=dup_of)
2795 if not user.is_staff:
2796 allowed_ids = permitted_document_ids(user, include_deleted=True)
2797 qs = qs.filter(pk__in=allowed_ids)
2798 return list(qs.values("id", "title", "deleted_at"))
2801class TaskSummarySerializer(serializers.Serializer[dict[str, Any]]):
2802 task_type = serializers.CharField()
2803 total_count = serializers.IntegerField()
2804 pending_count = serializers.IntegerField()
2805 success_count = serializers.IntegerField()
2806 failure_count = serializers.IntegerField()
2807 avg_duration_seconds = serializers.FloatField(allow_null=True)
2808 avg_wait_time_seconds = serializers.FloatField(allow_null=True)
2809 last_run = serializers.DateTimeField(allow_null=True)
2810 last_success = serializers.DateTimeField(allow_null=True)
2811 last_failure = serializers.DateTimeField(allow_null=True)
2814class RunTaskSerializer(serializers.Serializer[dict[str, str]]):
2815 task_type = serializers.ChoiceField(
2816 choices=PaperlessTask.TaskType.choices,
2817 label="Task Type",
2818 write_only=True,
2819 )
2822class AcknowledgeTasksViewSerializer(serializers.Serializer[dict[str, Any]]):
2823 tasks = serializers.ListField(
2824 required=False,
2825 label="Tasks",
2826 write_only=True,
2827 child=serializers.IntegerField(),
2828 )
2829 all = serializers.BooleanField(
2830 required=False,
2831 default=False,
2832 label="All",
2833 write_only=True,
2834 )
2836 def _validate_task_id_list(self, tasks, name="tasks") -> None:
2837 if not isinstance(tasks, list): 2837 ↛ 2838line 2837 didn't jump to line 2838 because the condition on line 2837 was never true
2838 raise serializers.ValidationError(f"{name} must be a list")
2839 if not all(isinstance(i, int) for i in tasks): 2839 ↛ 2840line 2839 didn't jump to line 2840 because the condition on line 2839 was never true
2840 raise serializers.ValidationError(f"{name} must be a list of integers")
2841 queryset = self.context.get("queryset", PaperlessTask.objects.all())
2842 count = queryset.filter(id__in=tasks).count()
2843 if not count == len(tasks):
2844 raise serializers.ValidationError(
2845 f"Some tasks in {name} don't exist or were specified twice.",
2846 )
2848 def validate_tasks(self, tasks):
2849 self._validate_task_id_list(tasks)
2850 return tasks
2852 def validate(self, attrs):
2853 acknowledge_all = attrs.get("all", False)
2854 task_ids = attrs.get("tasks")
2856 if acknowledge_all and task_ids is not None:
2857 raise serializers.ValidationError(
2858 "Set either all or tasks, not both.",
2859 )
2860 if not acknowledge_all and task_ids is None:
2861 raise serializers.ValidationError(
2862 "Either all must be true or tasks must be provided.",
2863 )
2865 return attrs
2868class ShareLinkSerializer(OwnedObjectSerializer):
2869 document_title = serializers.CharField(
2870 source="document.title",
2871 read_only=True,
2872 )
2874 class Meta:
2875 model = ShareLink
2876 fields = (
2877 "id",
2878 "created",
2879 "expiration",
2880 "slug",
2881 "document",
2882 "document_title",
2883 "file_version",
2884 )
2886 def create(self, validated_data):
2887 validated_data["slug"] = get_random_string(50)
2888 return super().create(validated_data)
2890 def validate_document(self, document):
2891 if (
2892 self.user is not None
2893 and self.user.has_perm("documents.view_document")
2894 and has_perms_owner_aware(
2895 self.user,
2896 "view_document",
2897 document,
2898 )
2899 ):
2900 return document
2901 raise PermissionDenied(
2902 _("Insufficient permissions."),
2903 )
2906class ShareLinkBundleSerializer(OwnedObjectSerializer):
2907 document_ids = serializers.ListField(
2908 child=serializers.IntegerField(min_value=1),
2909 allow_empty=False,
2910 write_only=True,
2911 )
2912 expiration_days = serializers.IntegerField(
2913 required=False,
2914 allow_null=True,
2915 min_value=1,
2916 write_only=True,
2917 )
2918 documents = serializers.PrimaryKeyRelatedField(
2919 many=True,
2920 read_only=True,
2921 )
2922 document_count = SerializerMethodField()
2924 class Meta:
2925 model = ShareLinkBundle
2926 fields = (
2927 "id",
2928 "created",
2929 "expiration",
2930 "expiration_days",
2931 "slug",
2932 "file_version",
2933 "status",
2934 "size_bytes",
2935 "last_error",
2936 "built_at",
2937 "documents",
2938 "document_ids",
2939 "document_count",
2940 )
2941 read_only_fields = (
2942 "id",
2943 "created",
2944 "expiration",
2945 "slug",
2946 "status",
2947 "size_bytes",
2948 "last_error",
2949 "built_at",
2950 "documents",
2951 "document_count",
2952 )
2954 def validate_document_ids(self, value):
2955 unique_ids = set(value)
2956 if len(unique_ids) != len(value):
2957 raise serializers.ValidationError(
2958 _("Duplicate document identifiers are not allowed."),
2959 )
2960 return value
2962 def create(self, validated_data):
2963 document_ids = validated_data.pop("document_ids")
2964 expiration_days = validated_data.pop("expiration_days", None)
2965 validated_data["slug"] = get_random_string(50)
2966 if expiration_days:
2967 validated_data["expiration"] = timezone.now() + timedelta(
2968 days=expiration_days,
2969 )
2970 else:
2971 validated_data["expiration"] = None
2973 share_link_bundle = super().create(validated_data)
2975 documents = list(
2976 Document.objects.filter(pk__in=document_ids).only(
2977 "pk",
2978 ),
2979 )
2980 documents_by_id = {doc.pk: doc for doc in documents}
2981 missing = [
2982 str(doc_id) for doc_id in document_ids if doc_id not in documents_by_id
2983 ]
2984 if missing:
2985 raise serializers.ValidationError(
2986 {
2987 "document_ids": _(
2988 "Documents not found: %(ids)s",
2989 )
2990 % {"ids": ", ".join(missing)},
2991 },
2992 )
2994 ordered_documents = [documents_by_id[doc_id] for doc_id in document_ids]
2995 share_link_bundle.documents.set(ordered_documents)
2996 share_link_bundle.document_total = len(ordered_documents)
2998 return share_link_bundle
3000 def get_document_count(self, obj: ShareLinkBundle) -> int:
3001 return getattr(obj, "document_total") or obj.documents.count()
3004class BulkEditObjectsSerializer(SerializerWithPerms, SetPermissionsMixin):
3005 objects = serializers.ListField(
3006 required=False,
3007 allow_empty=True,
3008 label="Objects",
3009 write_only=True,
3010 child=serializers.IntegerField(),
3011 )
3013 all = serializers.BooleanField(
3014 default=False,
3015 required=False,
3016 write_only=True,
3017 )
3019 filters = serializers.DictField(
3020 required=False,
3021 allow_empty=True,
3022 write_only=True,
3023 )
3025 object_type = serializers.ChoiceField(
3026 choices=[
3027 "tags",
3028 "correspondents",
3029 "document_types",
3030 "storage_paths",
3031 ],
3032 label="Object Type",
3033 write_only=True,
3034 )
3036 operation = serializers.ChoiceField(
3037 choices=[
3038 "set_permissions",
3039 "delete",
3040 ],
3041 label="Operation",
3042 required=True,
3043 write_only=True,
3044 )
3046 owner = serializers.PrimaryKeyRelatedField(
3047 queryset=User.objects.all(),
3048 required=False,
3049 allow_null=True,
3050 )
3052 permissions = SetPermissionsSerializer(
3053 label="Set permissions",
3054 required=False,
3055 write_only=True,
3056 )
3058 merge = serializers.BooleanField(
3059 default=False,
3060 write_only=True,
3061 required=False,
3062 )
3064 def get_object_class(self, object_type):
3065 object_class = None
3066 if object_type == "tags":
3067 object_class = Tag
3068 elif object_type == "correspondents":
3069 object_class = Correspondent
3070 elif object_type == "document_types":
3071 object_class = DocumentType
3072 elif object_type == "storage_paths": 3072 ↛ 3074line 3072 didn't jump to line 3074 because the condition on line 3072 was always true
3073 object_class = StoragePath
3074 return object_class
3076 def _validate_objects(self, objects, object_type):
3077 if not isinstance(objects, list): 3077 ↛ 3078line 3077 didn't jump to line 3078 because the condition on line 3077 was never true
3078 raise serializers.ValidationError("objects must be a list")
3079 if not all(isinstance(i, int) for i in objects): 3079 ↛ 3080line 3079 didn't jump to line 3080 because the condition on line 3079 was never true
3080 raise serializers.ValidationError("objects must be a list of integers")
3081 object_class = self.get_object_class(object_type)
3082 count = object_class.objects.filter(id__in=objects).count()
3083 if not count == len(objects): 3083 ↛ 3087line 3083 didn't jump to line 3087 because the condition on line 3083 was always true
3084 raise serializers.ValidationError(
3085 "Some ids in objects don't exist or were specified twice.",
3086 )
3087 return objects
3089 def _validate_permissions(self, permissions) -> dict:
3090 return self.validate_set_permissions(
3091 permissions,
3092 )
3094 def validate(self, attrs):
3095 object_type = attrs["object_type"]
3096 objects = attrs.get("objects")
3097 apply_to_all = attrs.get("all", False)
3098 operation = attrs.get("operation")
3100 if apply_to_all:
3101 attrs.setdefault("objects", [])
3102 else:
3103 if objects is None:
3104 raise serializers.ValidationError(
3105 "objects is required unless all is true.",
3106 )
3107 if len(objects) == 0:
3108 raise serializers.ValidationError("objects must not be empty")
3109 self._validate_objects(objects, object_type)
3111 if operation == "set_permissions":
3112 permissions = attrs.get("permissions")
3113 if permissions is not None: 3113 ↛ 3120line 3113 didn't jump to line 3120 because the condition on line 3113 was always true
3114 if not permissions: 3114 ↛ 3115line 3114 didn't jump to line 3115 because the condition on line 3114 was never true
3115 raise serializers.ValidationError(
3116 "permissions must not be empty",
3117 )
3118 attrs["permissions"] = self._validate_permissions(permissions)
3120 return attrs
3123class WorkflowTriggerSerializer(serializers.ModelSerializer[WorkflowTrigger]):
3124 id = serializers.IntegerField(required=False, allow_null=True)
3125 sources = fields.MultipleChoiceField(
3126 choices=WorkflowTrigger.DocumentSourceChoices.choices,
3127 allow_empty=True,
3128 default={
3129 DocumentSource.ConsumeFolder,
3130 DocumentSource.ApiUpload,
3131 DocumentSource.MailFetch,
3132 },
3133 )
3135 type = serializers.ChoiceField(
3136 choices=WorkflowTrigger.WorkflowTriggerType.choices,
3137 label="Trigger Type",
3138 )
3140 class Meta:
3141 model = WorkflowTrigger
3142 fields = [
3143 "id",
3144 "sources",
3145 "type",
3146 "filter_path",
3147 "filter_filename",
3148 "filter_mailrule",
3149 "matching_algorithm",
3150 "match",
3151 "is_insensitive",
3152 "filter_has_tags",
3153 "filter_has_all_tags",
3154 "filter_has_not_tags",
3155 "filter_custom_field_query",
3156 "filter_has_any_correspondents",
3157 "filter_has_not_correspondents",
3158 "filter_has_any_document_types",
3159 "filter_has_not_document_types",
3160 "filter_has_any_storage_paths",
3161 "filter_has_not_storage_paths",
3162 "filter_has_correspondent",
3163 "filter_has_document_type",
3164 "filter_has_storage_path",
3165 "schedule_offset_days",
3166 "schedule_is_recurring",
3167 "schedule_recurring_interval_days",
3168 "schedule_date_field",
3169 "schedule_date_custom_field",
3170 ]
3172 def validate(self, attrs):
3173 # Empty strings treated as None to avoid unexpected behavior
3174 if (
3175 "filter_filename" in attrs
3176 and attrs["filter_filename"] is not None
3177 and len(attrs["filter_filename"]) == 0
3178 ):
3179 attrs["filter_filename"] = None
3180 if (
3181 "filter_path" in attrs
3182 and attrs["filter_path"] is not None
3183 and len(attrs["filter_path"]) == 0
3184 ):
3185 attrs["filter_path"] = None
3187 if (
3188 "filter_custom_field_query" in attrs
3189 and attrs["filter_custom_field_query"] is not None
3190 and len(attrs["filter_custom_field_query"]) == 0
3191 ):
3192 attrs["filter_custom_field_query"] = None
3194 if (
3195 "filter_custom_field_query" in attrs
3196 and attrs["filter_custom_field_query"] is not None
3197 ):
3198 parser = CustomFieldQueryParser("filter_custom_field_query")
3199 parser.parse(attrs["filter_custom_field_query"])
3201 trigger_type = attrs.get("type", getattr(self.instance, "type", None))
3202 if (
3203 trigger_type == WorkflowTrigger.WorkflowTriggerType.CONSUMPTION
3204 and "filter_mailrule" not in attrs
3205 and ("filter_filename" not in attrs or attrs["filter_filename"] is None)
3206 and ("filter_path" not in attrs or attrs["filter_path"] is None)
3207 ):
3208 raise serializers.ValidationError(
3209 "File name, path or mail rule filter are required",
3210 )
3212 return attrs
3214 @staticmethod
3215 def normalize_workflow_trigger_sources(trigger) -> None:
3216 """
3217 Convert sources to strings to handle django-multiselectfield v1.0 changes
3218 """
3219 if trigger and "sources" in trigger:
3220 trigger["sources"] = [
3221 str(s.value if hasattr(s, "value") else s) for s in trigger["sources"]
3222 ]
3224 def create(self, validated_data):
3225 WorkflowTriggerSerializer.normalize_workflow_trigger_sources(validated_data)
3226 return super().create(validated_data)
3228 def update(self, instance, validated_data):
3229 WorkflowTriggerSerializer.normalize_workflow_trigger_sources(validated_data)
3230 return super().update(instance, validated_data)
3233class WorkflowActionEmailSerializer(serializers.ModelSerializer[WorkflowActionEmail]):
3234 id = serializers.IntegerField(allow_null=True, required=False)
3236 class Meta:
3237 model = WorkflowActionEmail
3238 fields = [
3239 "id",
3240 "subject",
3241 "body",
3242 "to",
3243 "include_document",
3244 ]
3247class WorkflowActionWebhookSerializer(
3248 serializers.ModelSerializer[WorkflowActionWebhook],
3249):
3250 id = serializers.IntegerField(allow_null=True, required=False)
3252 def validate_url(self, url):
3253 url_validator(url)
3254 return url
3256 class Meta:
3257 model = WorkflowActionWebhook
3258 fields = [
3259 "id",
3260 "url",
3261 "use_params",
3262 "as_json",
3263 "params",
3264 "body",
3265 "headers",
3266 "include_document",
3267 ]
3270class WorkflowActionSerializer(serializers.ModelSerializer[WorkflowAction]):
3271 id = serializers.IntegerField(required=False, allow_null=True)
3272 assign_correspondent = CorrespondentField(allow_null=True, required=False)
3273 assign_tags = TagsField(many=True, allow_null=True, required=False)
3274 assign_document_type = DocumentTypeField(allow_null=True, required=False)
3275 assign_storage_path = StoragePathField(allow_null=True, required=False)
3276 email = WorkflowActionEmailSerializer(allow_null=True, required=False)
3277 webhook = WorkflowActionWebhookSerializer(allow_null=True, required=False)
3279 class Meta:
3280 model = WorkflowAction
3281 fields = [
3282 "id",
3283 "type",
3284 "assign_title",
3285 "assign_tags",
3286 "assign_correspondent",
3287 "assign_document_type",
3288 "assign_storage_path",
3289 "assign_owner",
3290 "assign_view_users",
3291 "assign_view_groups",
3292 "assign_change_users",
3293 "assign_change_groups",
3294 "assign_custom_fields",
3295 "assign_custom_fields_values",
3296 "remove_all_tags",
3297 "remove_tags",
3298 "remove_all_correspondents",
3299 "remove_correspondents",
3300 "remove_all_document_types",
3301 "remove_document_types",
3302 "remove_all_storage_paths",
3303 "remove_storage_paths",
3304 "remove_custom_fields",
3305 "remove_all_custom_fields",
3306 "remove_all_owners",
3307 "remove_owners",
3308 "remove_all_permissions",
3309 "remove_view_users",
3310 "remove_view_groups",
3311 "remove_change_users",
3312 "remove_change_groups",
3313 "email",
3314 "webhook",
3315 "passwords",
3316 "ai_suggestion_fields",
3317 "ai_create_missing",
3318 "ai_overwrite_existing",
3319 ]
3321 def validate(self, attrs):
3322 if "assign_title" in attrs and attrs["assign_title"] is not None:
3323 if len(attrs["assign_title"]) == 0:
3324 # Empty strings treated as None to avoid unexpected behavior
3325 attrs["assign_title"] = None
3326 else:
3327 try:
3328 validate_workflow_template(attrs["assign_title"])
3329 except (ValueError, KeyError) as e:
3330 raise serializers.ValidationError(
3331 {"assign_title": f"{e.args[0]}"},
3332 )
3334 if attrs.get("assign_custom_fields_values"):
3335 # Empty strings treated as None to avoid unexpected behavior
3336 attrs["assign_custom_fields_values"] = {
3337 field_id: (None if value == "" else value)
3338 for field_id, value in attrs["assign_custom_fields_values"].items()
3339 }
3341 if (
3342 "type" in attrs
3343 and attrs["type"] == WorkflowAction.WorkflowActionType.EMAIL
3344 and "email" not in attrs
3345 ):
3346 raise serializers.ValidationError(
3347 "Email data is required for email actions",
3348 )
3350 if (
3351 "type" in attrs
3352 and attrs["type"] == WorkflowAction.WorkflowActionType.WEBHOOK
3353 and "webhook" not in attrs
3354 ):
3355 raise serializers.ValidationError(
3356 "Webhook data is required for webhook actions",
3357 )
3359 if (
3360 "type" in attrs
3361 and attrs["type"] == WorkflowAction.WorkflowActionType.PASSWORD_REMOVAL
3362 ):
3363 passwords = attrs.get("passwords")
3364 # ensure passwords is a non-empty list of non-empty strings
3365 if ( 3365 ↛ 3376line 3365 didn't jump to line 3376 because the condition on line 3365 was always true
3366 passwords is None
3367 or not isinstance(passwords, list)
3368 or len(passwords) == 0
3369 or any(not isinstance(pw, str) for pw in passwords)
3370 or any(len(pw.strip()) == 0 for pw in passwords)
3371 ):
3372 raise serializers.ValidationError(
3373 "Passwords are required for password removal actions",
3374 )
3376 if (
3377 "type" in attrs
3378 and attrs["type"] == WorkflowAction.WorkflowActionType.APPLY_AI_SUGGESTIONS
3379 ):
3380 fields = attrs.get("ai_suggestion_fields")
3381 valid_fields = set(WorkflowAction.AISuggestionField.values)
3382 if ( 3382 ↛ 3393line 3382 didn't jump to line 3393 because the condition on line 3382 was always true
3383 fields is None
3384 or not isinstance(fields, list)
3385 or len(fields) == 0
3386 or any(field not in valid_fields for field in fields)
3387 ):
3388 raise serializers.ValidationError(
3389 "At least one valid field is required for apply AI "
3390 f"suggestions actions, options are: {sorted(valid_fields)}",
3391 )
3393 return attrs
3396class WorkflowSerializer(serializers.ModelSerializer[Workflow]):
3397 order = serializers.IntegerField(required=False)
3399 triggers = WorkflowTriggerSerializer(many=True)
3400 actions = WorkflowActionSerializer(many=True)
3402 class Meta:
3403 model = Workflow
3404 fields = [
3405 "id",
3406 "name",
3407 "order",
3408 "enabled",
3409 "triggers",
3410 "actions",
3411 ]
3413 def validate(self, attrs):
3414 attrs = super().validate(attrs)
3416 if "actions" in attrs:
3417 has_remote_ocr_action = any(
3418 action.get("type") == WorkflowAction.WorkflowActionType.REMOTE_OCR
3419 for action in attrs["actions"]
3420 )
3421 has_ai_suggestions_action = any(
3422 action.get("type")
3423 == WorkflowAction.WorkflowActionType.APPLY_AI_SUGGESTIONS
3424 for action in attrs["actions"]
3425 )
3426 else:
3427 has_remote_ocr_action = self.instance is not None and (
3428 self.instance.actions.filter(
3429 type=WorkflowAction.WorkflowActionType.REMOTE_OCR,
3430 ).exists()
3431 )
3432 has_ai_suggestions_action = self.instance is not None and (
3433 self.instance.actions.filter(
3434 type=WorkflowAction.WorkflowActionType.APPLY_AI_SUGGESTIONS,
3435 ).exists()
3436 )
3438 if "triggers" in attrs:
3439 has_consumption_trigger = any(
3440 trigger.get("type") == WorkflowTrigger.WorkflowTriggerType.CONSUMPTION
3441 for trigger in attrs["triggers"]
3442 )
3443 has_non_consumption_trigger = any(
3444 trigger.get("type") != WorkflowTrigger.WorkflowTriggerType.CONSUMPTION
3445 for trigger in attrs["triggers"]
3446 )
3447 else:
3448 has_consumption_trigger = self.instance is not None and (
3449 self.instance.triggers.filter(
3450 type=WorkflowTrigger.WorkflowTriggerType.CONSUMPTION,
3451 ).exists()
3452 )
3453 has_non_consumption_trigger = self.instance is not None and (
3454 self.instance.triggers.exclude(
3455 type=WorkflowTrigger.WorkflowTriggerType.CONSUMPTION,
3456 ).exists()
3457 )
3459 # Remote OCR can only work with consumption triggers
3460 if has_remote_ocr_action and not has_consumption_trigger:
3461 raise serializers.ValidationError(
3462 "Remote OCR actions require a consumption started trigger",
3463 )
3465 # Suggestions are made from the document content, which does not exist
3466 # until after consumption has finished
3467 if has_ai_suggestions_action and not has_non_consumption_trigger: 3467 ↛ 3468line 3467 didn't jump to line 3468 because the condition on line 3467 was never true
3468 raise serializers.ValidationError(
3469 "Apply AI suggestions actions require a trigger other than "
3470 "consumption started",
3471 )
3473 return attrs
3475 def update_triggers_and_actions(
3476 self,
3477 instance: Workflow,
3478 triggers,
3479 actions,
3480 ) -> None:
3481 set_triggers = []
3482 set_actions = []
3484 if triggers is not None and triggers is not serializers.empty:
3485 for trigger in triggers:
3486 filter_has_tags = trigger.pop("filter_has_tags", None)
3487 filter_has_all_tags = trigger.pop("filter_has_all_tags", None)
3488 filter_has_not_tags = trigger.pop("filter_has_not_tags", None)
3489 filter_has_any_correspondents = trigger.pop(
3490 "filter_has_any_correspondents",
3491 None,
3492 )
3493 filter_has_not_correspondents = trigger.pop(
3494 "filter_has_not_correspondents",
3495 None,
3496 )
3497 filter_has_any_document_types = trigger.pop(
3498 "filter_has_any_document_types",
3499 None,
3500 )
3501 filter_has_not_document_types = trigger.pop(
3502 "filter_has_not_document_types",
3503 None,
3504 )
3505 filter_has_any_storage_paths = trigger.pop(
3506 "filter_has_any_storage_paths",
3507 None,
3508 )
3509 filter_has_not_storage_paths = trigger.pop(
3510 "filter_has_not_storage_paths",
3511 None,
3512 )
3513 # Convert sources to strings to handle django-multiselectfield v1.0 changes
3514 WorkflowTriggerSerializer.normalize_workflow_trigger_sources(trigger)
3515 trigger_instance, _ = WorkflowTrigger.objects.update_or_create(
3516 id=trigger.get("id"),
3517 defaults=trigger,
3518 )
3519 if filter_has_tags is not None:
3520 trigger_instance.filter_has_tags.set(filter_has_tags)
3521 if filter_has_all_tags is not None:
3522 trigger_instance.filter_has_all_tags.set(filter_has_all_tags)
3523 if filter_has_not_tags is not None:
3524 trigger_instance.filter_has_not_tags.set(filter_has_not_tags)
3525 if filter_has_any_correspondents is not None:
3526 trigger_instance.filter_has_any_correspondents.set(
3527 filter_has_any_correspondents,
3528 )
3529 if filter_has_not_correspondents is not None:
3530 trigger_instance.filter_has_not_correspondents.set(
3531 filter_has_not_correspondents,
3532 )
3533 if filter_has_any_document_types is not None:
3534 trigger_instance.filter_has_any_document_types.set(
3535 filter_has_any_document_types,
3536 )
3537 if filter_has_not_document_types is not None:
3538 trigger_instance.filter_has_not_document_types.set(
3539 filter_has_not_document_types,
3540 )
3541 if filter_has_any_storage_paths is not None:
3542 trigger_instance.filter_has_any_storage_paths.set(
3543 filter_has_any_storage_paths,
3544 )
3545 if filter_has_not_storage_paths is not None:
3546 trigger_instance.filter_has_not_storage_paths.set(
3547 filter_has_not_storage_paths,
3548 )
3549 set_triggers.append(trigger_instance)
3551 if actions is not None and actions is not serializers.empty:
3552 for index, action in enumerate(actions):
3553 action["order"] = index
3554 assign_tags = action.pop("assign_tags", None)
3555 assign_view_users = action.pop("assign_view_users", None)
3556 assign_view_groups = action.pop("assign_view_groups", None)
3557 assign_change_users = action.pop("assign_change_users", None)
3558 assign_change_groups = action.pop("assign_change_groups", None)
3559 assign_custom_fields = action.pop("assign_custom_fields", None)
3560 remove_tags = action.pop("remove_tags", None)
3561 remove_correspondents = action.pop("remove_correspondents", None)
3562 remove_document_types = action.pop("remove_document_types", None)
3563 remove_storage_paths = action.pop("remove_storage_paths", None)
3564 remove_custom_fields = action.pop("remove_custom_fields", None)
3565 remove_owners = action.pop("remove_owners", None)
3566 remove_view_users = action.pop("remove_view_users", None)
3567 remove_view_groups = action.pop("remove_view_groups", None)
3568 remove_change_users = action.pop("remove_change_users", None)
3569 remove_change_groups = action.pop("remove_change_groups", None)
3571 email_data = action.pop("email", None)
3572 webhook_data = action.pop("webhook", None)
3574 action_instance, _ = WorkflowAction.objects.update_or_create(
3575 id=action.get("id"),
3576 defaults=action,
3577 )
3579 if email_data is not None:
3580 serializer = WorkflowActionEmailSerializer(data=email_data)
3581 serializer.is_valid(raise_exception=True)
3582 email, _ = WorkflowActionEmail.objects.update_or_create(
3583 id=email_data.get("id"),
3584 defaults=serializer.validated_data,
3585 )
3586 action_instance.email = email
3587 action_instance.save()
3589 if webhook_data is not None: 3589 ↛ 3590line 3589 didn't jump to line 3590 because the condition on line 3589 was never true
3590 serializer = WorkflowActionWebhookSerializer(data=webhook_data)
3591 serializer.is_valid(raise_exception=True)
3592 webhook, _ = WorkflowActionWebhook.objects.update_or_create(
3593 id=webhook_data.get("id"),
3594 defaults=serializer.validated_data,
3595 )
3596 action_instance.webhook = webhook
3597 action_instance.save()
3599 if assign_tags is not None:
3600 action_instance.assign_tags.set(assign_tags)
3601 if assign_view_users is not None:
3602 action_instance.assign_view_users.set(assign_view_users)
3603 if assign_view_groups is not None:
3604 action_instance.assign_view_groups.set(assign_view_groups)
3605 if assign_change_users is not None:
3606 action_instance.assign_change_users.set(assign_change_users)
3607 if assign_change_groups is not None:
3608 action_instance.assign_change_groups.set(assign_change_groups)
3609 if assign_custom_fields is not None:
3610 action_instance.assign_custom_fields.set(assign_custom_fields)
3611 if remove_tags is not None:
3612 action_instance.remove_tags.set(remove_tags)
3613 if remove_correspondents is not None:
3614 action_instance.remove_correspondents.set(remove_correspondents)
3615 if remove_document_types is not None:
3616 action_instance.remove_document_types.set(remove_document_types)
3617 if remove_storage_paths is not None:
3618 action_instance.remove_storage_paths.set(remove_storage_paths)
3619 if remove_custom_fields is not None:
3620 action_instance.remove_custom_fields.set(remove_custom_fields)
3621 if remove_owners is not None:
3622 action_instance.remove_owners.set(remove_owners)
3623 if remove_view_users is not None:
3624 action_instance.remove_view_users.set(remove_view_users)
3625 if remove_view_groups is not None:
3626 action_instance.remove_view_groups.set(remove_view_groups)
3627 if remove_change_users is not None:
3628 action_instance.remove_change_users.set(remove_change_users)
3629 if remove_change_groups is not None:
3630 action_instance.remove_change_groups.set(remove_change_groups)
3632 set_actions.append(action_instance)
3634 if triggers is not serializers.empty:
3635 instance.triggers.set(set_triggers)
3636 if actions is not serializers.empty:
3637 instance.actions.set(set_actions)
3638 instance.save()
3640 def prune_triggers_and_actions(self) -> None:
3641 """
3642 ManyToMany fields dont support e.g. on_delete so we need to discard unattached
3643 triggers and actions manually
3644 """
3645 WorkflowTrigger.objects.annotate(
3646 workflow_count=Count("workflows"),
3647 ).filter(workflow_count=0).delete()
3649 WorkflowAction.objects.annotate(
3650 workflow_count=Count("workflows"),
3651 ).filter(workflow_count=0).delete()
3653 WorkflowActionEmail.objects.filter(action=None).delete()
3654 WorkflowActionWebhook.objects.filter(action=None).delete()
3656 def create(self, validated_data) -> Workflow:
3657 if "triggers" in validated_data: 3657 ↛ 3660line 3657 didn't jump to line 3660 because the condition on line 3657 was always true
3658 triggers = validated_data.pop("triggers")
3660 if "actions" in validated_data: 3660 ↛ 3665line 3660 didn't jump to line 3665 because the condition on line 3660 was always true
3661 actions = validated_data.pop("actions")
3662 for action in actions:
3663 action.pop("id", None)
3665 instance = super().create(validated_data)
3667 self.update_triggers_and_actions(instance, triggers, actions)
3669 return instance
3671 def update(self, instance: Workflow, validated_data) -> Workflow:
3672 triggers = validated_data.pop("triggers", serializers.empty)
3673 actions = validated_data.pop("actions", serializers.empty)
3675 instance = super().update(instance, validated_data)
3677 self.update_triggers_and_actions(instance, triggers, actions)
3678 self.prune_triggers_and_actions()
3680 return instance
3683class TrashSerializer(SerializerWithPerms):
3684 documents = serializers.ListField(
3685 required=False,
3686 label="Documents",
3687 write_only=True,
3688 child=serializers.IntegerField(),
3689 )
3691 action = serializers.ChoiceField(
3692 choices=["restore", "empty"],
3693 label="Action",
3694 write_only=True,
3695 )
3697 def validate_documents(self, documents: list[int]) -> list[int]:
3698 count = Document.deleted_objects.filter(id__in=documents).count()
3699 if not count == len(documents):
3700 raise serializers.ValidationError(
3701 "Some documents in the list have not yet been deleted.",
3702 )
3703 return documents
3706class StoragePathTestSerializer(SerializerWithPerms):
3707 path = serializers.CharField(
3708 required=True,
3709 label="Path",
3710 write_only=True,
3711 )
3713 document = serializers.PrimaryKeyRelatedField(
3714 queryset=Document.objects.none(),
3715 required=True,
3716 label="Document",
3717 write_only=True,
3718 )
3720 def __init__(self, *args: Any, **kwargs: Any) -> None:
3721 super().__init__(*args, **kwargs)
3722 request = self.context.get("request")
3723 user = getattr(request, "user", None) if request else None
3724 if user is not None and user.is_authenticated:
3725 document_field = self.fields.get("document")
3726 if not isinstance(document_field, serializers.PrimaryKeyRelatedField): 3726 ↛ 3727line 3726 didn't jump to line 3727 because the condition on line 3726 was never true
3727 return
3728 document_field.queryset = Document.objects.filter(
3729 id__in=permitted_document_ids(user),
3730 )