mirror of
https://github.com/paperless-ngx/paperless-ngx.git
synced 2026-09-28 04:10:32 +00:00
Refactor: extract serialisers.py into a structured module
documents/serialisers.py had grown into a single file with every serialiser for the app, making it hard to navigate and putting unrelated domains behind one import path. Moves serialisers into documents/serialisers/ as a package, grouping them into logical per-domain files instead. paperless_mail/serialisers.py is updated to import from the new submodules.
This commit is contained in:
stumpylog
File diff suppressed because it is too large
Load Diff
@@ -0,0 +1,568 @@
|
||||
from __future__ import annotations
|
||||
|
||||
import logging
|
||||
from typing import TYPE_CHECKING
|
||||
from typing import Any
|
||||
from typing import Literal
|
||||
|
||||
from django.contrib.auth.models import Group
|
||||
from django.contrib.auth.models import User
|
||||
from django.contrib.contenttypes.models import ContentType
|
||||
from django.utils.text import slugify
|
||||
from django.utils.translation import gettext as _
|
||||
from drf_spectacular.utils import extend_schema_field
|
||||
from guardian.core import ObjectPermissionChecker
|
||||
from guardian.shortcuts import get_users_with_perms
|
||||
from guardian.utils import get_group_obj_perms_model
|
||||
from guardian.utils import get_user_obj_perms_model
|
||||
from rest_framework import serializers
|
||||
from rest_framework.exceptions import PermissionDenied
|
||||
from rest_framework.fields import SerializerMethodField
|
||||
from rest_framework.utils import model_meta
|
||||
|
||||
from documents import bulk_edit
|
||||
from documents.models import Document
|
||||
from documents.models import MatchingModel
|
||||
from documents.models import Note
|
||||
from documents.permissions import get_groups_with_only_permission
|
||||
from documents.permissions import set_permissions_for_object
|
||||
from documents.regex import validate_regex_pattern
|
||||
|
||||
if TYPE_CHECKING:
|
||||
from collections.abc import Iterable
|
||||
|
||||
|
||||
logger = logging.getLogger("paperless.serializers")
|
||||
|
||||
|
||||
# https://www.django-rest-framework.org/api-guide/serializers/#example
|
||||
class DynamicFieldsModelSerializer(serializers.ModelSerializer[Any]):
|
||||
"""
|
||||
A ModelSerializer that takes an additional `fields` argument that
|
||||
controls which fields should be displayed.
|
||||
"""
|
||||
|
||||
def __init__(self, *args, **kwargs) -> None:
|
||||
# Don't pass the 'fields' arg up to the superclass
|
||||
fields = kwargs.pop("fields", None)
|
||||
|
||||
# Instantiate the superclass normally
|
||||
super().__init__(*args, **kwargs)
|
||||
|
||||
if fields is not None:
|
||||
# Drop any fields that are not specified in the `fields` argument.
|
||||
allowed = set(fields)
|
||||
existing = set(self.fields)
|
||||
for field_name in existing - allowed:
|
||||
self.fields.pop(field_name)
|
||||
|
||||
|
||||
class DocumentUpdateFieldsModelSerializer(DynamicFieldsModelSerializer):
|
||||
stale_update_excluded_fields = frozenset({"filename", "archive_filename"})
|
||||
|
||||
def _get_update_fields(self, validated_data) -> list[str]:
|
||||
model_fields = {
|
||||
field.name
|
||||
for field in self.Meta.model._meta.concrete_fields
|
||||
if field.name not in self.stale_update_excluded_fields
|
||||
}
|
||||
update_fields = [
|
||||
field_name for field_name in validated_data if field_name in model_fields
|
||||
]
|
||||
if "modified" in model_fields and "modified" not in update_fields:
|
||||
update_fields.append("modified")
|
||||
return update_fields
|
||||
|
||||
def update(self, instance, validated_data):
|
||||
serializers.raise_errors_on_nested_writes("update", self, validated_data)
|
||||
info = model_meta.get_field_info(instance)
|
||||
|
||||
m2m_fields = []
|
||||
for attr, value in validated_data.items():
|
||||
if attr in info.relations and info.relations[attr].to_many:
|
||||
m2m_fields.append((attr, value))
|
||||
else:
|
||||
setattr(instance, attr, value)
|
||||
|
||||
# File names are managed by post-save file handling. Saving only the
|
||||
# serializer-updated fields prevents stale in-memory path values from
|
||||
# overwriting a concurrent move.
|
||||
instance.save(update_fields=self._get_update_fields(validated_data))
|
||||
|
||||
for attr, value in m2m_fields:
|
||||
field = getattr(instance, attr)
|
||||
field.set(value)
|
||||
|
||||
return instance
|
||||
|
||||
|
||||
class MatchingModelSerializer(serializers.ModelSerializer[Any]):
|
||||
document_count = serializers.IntegerField(read_only=True)
|
||||
|
||||
def get_slug(self, obj) -> str:
|
||||
return slugify(obj.name)
|
||||
|
||||
slug = SerializerMethodField()
|
||||
|
||||
def validate(self, data):
|
||||
# TODO: remove pending https://github.com/encode/django-rest-framework/issues/7173
|
||||
name = data.get(
|
||||
"name",
|
||||
self.instance.name if hasattr(self.instance, "name") else None,
|
||||
)
|
||||
owner = (
|
||||
data["owner"]
|
||||
if "owner" in data
|
||||
else self.user
|
||||
if hasattr(self, "user")
|
||||
else None
|
||||
)
|
||||
pk = self.instance.pk if hasattr(self.instance, "pk") else None
|
||||
if ("name" in data or "owner" in data) and self.Meta.model.objects.filter(
|
||||
name=name,
|
||||
owner=owner,
|
||||
).exclude(pk=pk).exists():
|
||||
raise serializers.ValidationError(
|
||||
{"error": "Object violates owner / name unique constraint"},
|
||||
)
|
||||
return data
|
||||
|
||||
def validate_match(self, match):
|
||||
if (
|
||||
"matching_algorithm" in self.initial_data
|
||||
and self.initial_data["matching_algorithm"] == MatchingModel.MATCH_REGEX
|
||||
):
|
||||
try:
|
||||
validate_regex_pattern(match)
|
||||
except ValueError as e:
|
||||
logger.debug(f"Invalid regular expression: {e!s}")
|
||||
raise serializers.ValidationError(
|
||||
"Invalid regular expression, see log for details.",
|
||||
)
|
||||
return match
|
||||
|
||||
|
||||
PERMISSION_ACTIONS = ("view", "change")
|
||||
|
||||
|
||||
class SetPermissionsMixin:
|
||||
def _validate_user_ids(self, user_ids):
|
||||
users = User.objects.none()
|
||||
if user_ids is not None:
|
||||
users = User.objects.filter(id__in=user_ids)
|
||||
if not users.count() == len(user_ids):
|
||||
raise serializers.ValidationError(
|
||||
"Some users in don't exist or were specified twice.",
|
||||
)
|
||||
return users
|
||||
|
||||
def _validate_group_ids(self, group_ids):
|
||||
groups = Group.objects.none()
|
||||
if group_ids is not None:
|
||||
groups = Group.objects.filter(id__in=group_ids)
|
||||
if not groups.count() == len(group_ids):
|
||||
raise serializers.ValidationError(
|
||||
"Some groups in don't exist or were specified twice.",
|
||||
)
|
||||
return groups
|
||||
|
||||
def validate_set_permissions(self, set_permissions=None):
|
||||
permissions_dict = {action: {} for action in PERMISSION_ACTIONS}
|
||||
if set_permissions is not None:
|
||||
for action in PERMISSION_ACTIONS:
|
||||
if action in set_permissions:
|
||||
if "users" in set_permissions[action]:
|
||||
users = set_permissions[action]["users"]
|
||||
permissions_dict[action]["users"] = self._validate_user_ids(
|
||||
users,
|
||||
)
|
||||
if "groups" in set_permissions[action]:
|
||||
groups = set_permissions[action]["groups"]
|
||||
permissions_dict[action]["groups"] = self._validate_group_ids(
|
||||
groups,
|
||||
)
|
||||
else:
|
||||
del permissions_dict[action]
|
||||
return permissions_dict
|
||||
|
||||
def _set_permissions(self, permissions, object) -> None:
|
||||
set_permissions_for_object(permissions, object)
|
||||
|
||||
|
||||
class SerializerWithPerms(serializers.Serializer[dict[str, Any]]):
|
||||
def __init__(self, *args, **kwargs) -> None:
|
||||
self.user = kwargs.pop("user", None)
|
||||
self.full_perms = kwargs.pop("full_perms", False)
|
||||
self.all_fields = kwargs.pop("all_fields", False)
|
||||
super().__init__(*args, **kwargs)
|
||||
|
||||
|
||||
class PermissionSetSerializer(serializers.Serializer[dict[str, Any]]):
|
||||
users = serializers.ListField(
|
||||
child=serializers.IntegerField(),
|
||||
required=False,
|
||||
allow_null=True,
|
||||
)
|
||||
groups = serializers.ListField(
|
||||
child=serializers.IntegerField(),
|
||||
required=False,
|
||||
allow_null=True,
|
||||
)
|
||||
|
||||
|
||||
class SetPermissionsSerializer(serializers.Serializer[dict[str, Any]]):
|
||||
view = PermissionSetSerializer(required=False)
|
||||
change = PermissionSetSerializer(required=False)
|
||||
|
||||
def to_internal_value(self, data):
|
||||
if isinstance(data, dict):
|
||||
unknown_keys = set(data) - set(PERMISSION_ACTIONS)
|
||||
if unknown_keys:
|
||||
raise serializers.ValidationError(
|
||||
{key: "Unknown permission action." for key in sorted(unknown_keys)},
|
||||
)
|
||||
return super().to_internal_value(data)
|
||||
|
||||
|
||||
class OwnedObjectSerializer(
|
||||
SerializerWithPerms,
|
||||
serializers.ModelSerializer[Any],
|
||||
SetPermissionsMixin,
|
||||
):
|
||||
def __init__(self, *args, **kwargs) -> None:
|
||||
super().__init__(*args, **kwargs)
|
||||
|
||||
if not self.all_fields:
|
||||
try:
|
||||
if self.full_perms:
|
||||
self.fields.pop("user_can_change")
|
||||
self.fields.pop("is_shared_by_requester")
|
||||
else:
|
||||
self.fields.pop("permissions")
|
||||
except KeyError:
|
||||
pass
|
||||
|
||||
def _get_perms(self, obj, codename: str, target: Literal["users", "groups"]):
|
||||
"""
|
||||
Get the given permissions from context or from django-guardian.
|
||||
|
||||
:param codename: The permission codename, e.g. 'view' or 'change'
|
||||
:param target: 'users' or 'groups'
|
||||
"""
|
||||
key = f"{target}_{codename}_perms"
|
||||
cached = self.context.get(key, {}).get(obj.pk)
|
||||
if cached is not None:
|
||||
return list(cached)
|
||||
|
||||
# Permission not found in the context, get it from guardian
|
||||
if target == "users":
|
||||
return list(
|
||||
get_users_with_perms(
|
||||
obj,
|
||||
only_with_perms_in=[f"{codename}_{obj.__class__.__name__.lower()}"],
|
||||
with_group_users=False,
|
||||
).values_list("id", flat=True),
|
||||
)
|
||||
else: # groups
|
||||
return list(
|
||||
get_groups_with_only_permission(
|
||||
obj,
|
||||
codename=f"{codename}_{obj.__class__.__name__.lower()}",
|
||||
).values_list("id", flat=True),
|
||||
)
|
||||
|
||||
@extend_schema_field(
|
||||
field={
|
||||
"type": "object",
|
||||
"properties": {
|
||||
"view": {
|
||||
"type": "object",
|
||||
"properties": {
|
||||
"users": {
|
||||
"type": "array",
|
||||
"items": {"type": "integer"},
|
||||
},
|
||||
"groups": {
|
||||
"type": "array",
|
||||
"items": {"type": "integer"},
|
||||
},
|
||||
},
|
||||
},
|
||||
"change": {
|
||||
"type": "object",
|
||||
"properties": {
|
||||
"users": {
|
||||
"type": "array",
|
||||
"items": {"type": "integer"},
|
||||
},
|
||||
"groups": {
|
||||
"type": "array",
|
||||
"items": {"type": "integer"},
|
||||
},
|
||||
},
|
||||
},
|
||||
},
|
||||
},
|
||||
)
|
||||
def get_permissions(self, obj) -> dict:
|
||||
return {
|
||||
"view": {
|
||||
"users": self._get_perms(obj, "view", "users"),
|
||||
"groups": self._get_perms(obj, "view", "groups"),
|
||||
},
|
||||
"change": {
|
||||
"users": self._get_perms(obj, "change", "users"),
|
||||
"groups": self._get_perms(obj, "change", "groups"),
|
||||
},
|
||||
}
|
||||
|
||||
def get_user_can_change(self, obj) -> bool:
|
||||
if obj.owner is None or obj.owner == self.user:
|
||||
return True
|
||||
if self.user is None:
|
||||
return False
|
||||
if self.user.is_active and self.user.is_superuser:
|
||||
# Mirrors guardian's own ObjectPermissionChecker.has_perm() shortcut --
|
||||
# superusers aren't necessarily granted explicit object permissions,
|
||||
# so the batched context below would otherwise incorrectly say no.
|
||||
return True
|
||||
|
||||
# Prefer the page-level batch computed by BulkPermissionMixin
|
||||
# (get_serializer_context) over a fresh per-object guardian check,
|
||||
# which would otherwise query the permission tables once per row.
|
||||
users_change_perms = self.context.get("users_change_perms")
|
||||
groups_change_perms = self.context.get("groups_change_perms")
|
||||
if users_change_perms is not None and groups_change_perms is not None:
|
||||
if self.user.pk in users_change_perms.get(obj.pk, []):
|
||||
return True
|
||||
user_group_ids = getattr(self, "_user_group_ids", None)
|
||||
if user_group_ids is None:
|
||||
user_group_ids = set(self.user.groups.values_list("id", flat=True))
|
||||
self._user_group_ids = user_group_ids
|
||||
return bool(
|
||||
user_group_ids.intersection(groups_change_perms.get(obj.pk, [])),
|
||||
)
|
||||
|
||||
checker = ObjectPermissionChecker(self.user)
|
||||
return checker.has_perm(f"change_{obj.__class__.__name__.lower()}", obj)
|
||||
|
||||
@staticmethod
|
||||
def get_shared_object_pks(objects: Iterable):
|
||||
"""
|
||||
Return the primary keys of the subset of objects that are shared.
|
||||
"""
|
||||
try:
|
||||
first_obj = next(iter(objects))
|
||||
except StopIteration:
|
||||
return set()
|
||||
|
||||
ctype = ContentType.objects.get_for_model(first_obj)
|
||||
object_pks = list(obj.pk for obj in objects)
|
||||
pk_type = type(first_obj.pk)
|
||||
|
||||
def get_pks_for_permission_type(model):
|
||||
return map(
|
||||
pk_type, # coerce the pk to be the same type of the provided objects
|
||||
model.objects.filter(
|
||||
content_type=ctype,
|
||||
object_pk__in=object_pks,
|
||||
)
|
||||
.values_list("object_pk", flat=True)
|
||||
.distinct(),
|
||||
)
|
||||
|
||||
UserObjectPermission = get_user_obj_perms_model()
|
||||
GroupObjectPermission = get_group_obj_perms_model()
|
||||
user_permission_pks = get_pks_for_permission_type(UserObjectPermission)
|
||||
group_permission_pks = get_pks_for_permission_type(GroupObjectPermission)
|
||||
|
||||
return set(user_permission_pks) | set(group_permission_pks)
|
||||
|
||||
def get_is_shared_by_requester(self, obj: Document) -> bool:
|
||||
# First check the context to see if `shared_object_pks` is set by the parent.
|
||||
shared_object_pks = self.context.get("shared_object_pks")
|
||||
# If not just check if the current object is shared.
|
||||
if shared_object_pks is None:
|
||||
shared_object_pks = self.get_shared_object_pks([obj])
|
||||
return obj.owner == self.user and obj.id in shared_object_pks
|
||||
|
||||
permissions = SerializerMethodField(read_only=True, required=False)
|
||||
user_can_change = SerializerMethodField(read_only=True, required=False)
|
||||
is_shared_by_requester = SerializerMethodField(read_only=True, required=False)
|
||||
|
||||
set_permissions = SetPermissionsSerializer(
|
||||
label="Set permissions",
|
||||
required=False,
|
||||
write_only=True,
|
||||
)
|
||||
# other methods in mixin
|
||||
|
||||
def validate_unique_together(self, validated_data, instance=None) -> None:
|
||||
# workaround for https://github.com/encode/django-rest-framework/issues/9358
|
||||
if "owner" in validated_data and "name" in self.Meta.fields:
|
||||
name = validated_data.get("name", instance.name if instance else None)
|
||||
objects = (
|
||||
self.Meta.model.objects.exclude(pk=instance.pk)
|
||||
if instance
|
||||
else self.Meta.model.objects.all()
|
||||
)
|
||||
not_unique = objects.filter(
|
||||
owner=validated_data["owner"],
|
||||
name=name,
|
||||
).exists()
|
||||
if not_unique:
|
||||
raise serializers.ValidationError(
|
||||
{"error": "Object violates owner / name unique constraint"},
|
||||
)
|
||||
|
||||
def create(self, validated_data):
|
||||
# default to current user if not set
|
||||
request = self.context.get("request")
|
||||
if (
|
||||
"owner" not in validated_data
|
||||
or (request is not None and "owner" not in request.data)
|
||||
) and self.user:
|
||||
validated_data["owner"] = self.user
|
||||
permissions = None
|
||||
if "set_permissions" in validated_data:
|
||||
permissions = validated_data.pop("set_permissions")
|
||||
self.validate_unique_together(validated_data)
|
||||
instance = super().create(validated_data)
|
||||
if permissions is not None:
|
||||
self._set_permissions(permissions, instance)
|
||||
return instance
|
||||
|
||||
def update(self, instance, validated_data):
|
||||
user = getattr(self, "user", None)
|
||||
is_superuser = user.is_superuser if user is not None else False
|
||||
is_owner = instance.owner == user if user is not None else False
|
||||
is_unowned = instance.owner is None
|
||||
|
||||
if (
|
||||
("owner" in validated_data and validated_data["owner"] != instance.owner)
|
||||
or "set_permissions" in validated_data
|
||||
) and not (is_superuser or is_owner or is_unowned):
|
||||
raise PermissionDenied(
|
||||
_("Insufficient permissions."),
|
||||
)
|
||||
|
||||
if "set_permissions" in validated_data:
|
||||
self._set_permissions(validated_data["set_permissions"], instance)
|
||||
self.validate_unique_together(validated_data, instance)
|
||||
return super().update(instance, validated_data)
|
||||
|
||||
|
||||
class OwnedObjectListSerializer(serializers.ListSerializer[Any]):
|
||||
def to_representation(self, documents):
|
||||
self.child.context["shared_object_pks"] = self.child.get_shared_object_pks(
|
||||
documents,
|
||||
)
|
||||
return super().to_representation(documents)
|
||||
|
||||
|
||||
class ReadWriteSerializerMethodField(serializers.SerializerMethodField):
|
||||
"""
|
||||
Based on https://stackoverflow.com/a/62579804
|
||||
"""
|
||||
|
||||
def __init__(self, method_name=None, *args, **kwargs) -> None:
|
||||
self.method_name = method_name
|
||||
kwargs["source"] = "*"
|
||||
super(serializers.SerializerMethodField, self).__init__(*args, **kwargs)
|
||||
|
||||
def to_internal_value(self, data):
|
||||
return {self.field_name: data}
|
||||
|
||||
|
||||
class DocumentListSerializer(serializers.Serializer[dict[str, list[int]]]):
|
||||
documents = serializers.ListField(
|
||||
required=True,
|
||||
label="Documents",
|
||||
write_only=True,
|
||||
child=serializers.IntegerField(),
|
||||
)
|
||||
|
||||
def _validate_document_id_list(self, documents, name="documents") -> None:
|
||||
if not isinstance(documents, list):
|
||||
raise serializers.ValidationError(f"{name} must be a list")
|
||||
if not all(isinstance(i, int) for i in documents):
|
||||
raise serializers.ValidationError(f"{name} must be a list of integers")
|
||||
count = Document.objects.filter(id__in=documents).count()
|
||||
if not count == len(documents):
|
||||
raise serializers.ValidationError(
|
||||
f"Some documents in {name} don't exist or were specified twice.",
|
||||
)
|
||||
|
||||
def validate_documents(self, documents):
|
||||
self._validate_document_id_list(documents)
|
||||
return documents
|
||||
|
||||
|
||||
class DocumentSelectionSerializer(DocumentListSerializer):
|
||||
documents = serializers.ListField(
|
||||
required=False,
|
||||
label="Documents",
|
||||
write_only=True,
|
||||
child=serializers.IntegerField(),
|
||||
)
|
||||
|
||||
all = serializers.BooleanField(
|
||||
default=False,
|
||||
required=False,
|
||||
write_only=True,
|
||||
)
|
||||
|
||||
filters = serializers.DictField(
|
||||
required=False,
|
||||
allow_empty=True,
|
||||
write_only=True,
|
||||
)
|
||||
|
||||
excluded_documents = serializers.ListField(
|
||||
required=False,
|
||||
default=list,
|
||||
write_only=True,
|
||||
child=serializers.IntegerField(),
|
||||
)
|
||||
|
||||
def validate(self, attrs):
|
||||
if attrs.get("all", False):
|
||||
attrs.setdefault("documents", [])
|
||||
return attrs
|
||||
|
||||
if attrs["excluded_documents"]:
|
||||
raise serializers.ValidationError(
|
||||
"excluded_documents is only supported when all is true.",
|
||||
)
|
||||
|
||||
if "documents" not in attrs:
|
||||
raise serializers.ValidationError(
|
||||
"documents is required unless all is true.",
|
||||
)
|
||||
|
||||
documents = attrs["documents"]
|
||||
self._validate_document_id_list(documents)
|
||||
return attrs
|
||||
|
||||
|
||||
class SourceModeValidationMixin:
|
||||
def validate_source_mode(self, source_mode: str) -> str:
|
||||
if source_mode not in bulk_edit.SourceModeChoices.__dict__.values():
|
||||
raise serializers.ValidationError("Invalid source_mode")
|
||||
return source_mode
|
||||
|
||||
|
||||
class BasicUserSerializer(serializers.ModelSerializer[User]):
|
||||
# Different than paperless.serializers.UserSerializer
|
||||
class Meta:
|
||||
model = User
|
||||
fields = ["id", "username", "first_name", "last_name"]
|
||||
|
||||
|
||||
class NotesSerializer(serializers.ModelSerializer[Note]):
|
||||
user = BasicUserSerializer(read_only=True)
|
||||
|
||||
class Meta:
|
||||
model = Note
|
||||
fields = ["id", "note", "created", "user"]
|
||||
ordering = ["-created"]
|
||||
@@ -0,0 +1,750 @@
|
||||
from __future__ import annotations
|
||||
|
||||
import logging
|
||||
|
||||
from django.contrib.auth.models import User
|
||||
from rest_framework import serializers
|
||||
|
||||
from documents import bulk_edit
|
||||
from documents.models import Correspondent
|
||||
from documents.models import CustomField
|
||||
from documents.models import Document
|
||||
from documents.models import DocumentType
|
||||
from documents.models import StoragePath
|
||||
from documents.models import Tag
|
||||
|
||||
from .base import DocumentListSerializer
|
||||
from .base import DocumentSelectionSerializer
|
||||
from .base import SerializerWithPerms
|
||||
from .base import SetPermissionsMixin
|
||||
from .base import SetPermissionsSerializer
|
||||
from .base import SourceModeValidationMixin
|
||||
from .metadata import CustomFieldInstanceSerializer
|
||||
|
||||
logger = logging.getLogger("paperless.serializers")
|
||||
|
||||
|
||||
def _validate_rotation_degrees(degrees: int, field: str = "degrees") -> int:
|
||||
# QPDF refuses any other angle, which would otherwise fail inside the task
|
||||
if degrees % 90 != 0:
|
||||
raise serializers.ValidationError(f"{field} must be a multiple of 90")
|
||||
return degrees
|
||||
|
||||
|
||||
class RotateDocumentsSerializer(DocumentSelectionSerializer, SourceModeValidationMixin):
|
||||
degrees = serializers.IntegerField(required=True)
|
||||
source_mode = serializers.CharField(
|
||||
required=False,
|
||||
default=bulk_edit.SourceModeChoices.LATEST_VERSION,
|
||||
)
|
||||
from_webui = serializers.BooleanField(required=False, default=False)
|
||||
|
||||
def validate_degrees(self, value: int) -> int:
|
||||
return _validate_rotation_degrees(value)
|
||||
|
||||
|
||||
class MergeDocumentsSerializer(DocumentListSerializer, SourceModeValidationMixin):
|
||||
metadata_document_id = serializers.IntegerField(
|
||||
required=False,
|
||||
allow_null=True,
|
||||
)
|
||||
delete_originals = serializers.BooleanField(required=False, default=False)
|
||||
archive_fallback = serializers.BooleanField(required=False, default=False)
|
||||
source_mode = serializers.CharField(
|
||||
required=False,
|
||||
default=bulk_edit.SourceModeChoices.LATEST_VERSION,
|
||||
)
|
||||
from_webui = serializers.BooleanField(required=False, default=False)
|
||||
|
||||
|
||||
class MergeDocumentsAsVersionsSerializer(DocumentListSerializer):
|
||||
root_document_id = serializers.IntegerField(required=True)
|
||||
version_label = serializers.CharField(
|
||||
required=False,
|
||||
allow_blank=True,
|
||||
allow_null=True,
|
||||
max_length=64,
|
||||
)
|
||||
|
||||
def validate_version_label(self, value):
|
||||
if value is None:
|
||||
return None
|
||||
normalized = value.strip()
|
||||
return normalized or None
|
||||
|
||||
def validate(self, attrs):
|
||||
documents = attrs["documents"]
|
||||
if len(documents) < 2:
|
||||
raise serializers.ValidationError(
|
||||
"At least two documents are required.",
|
||||
)
|
||||
if attrs.get("version_label") is not None and len(documents) != 2:
|
||||
raise serializers.ValidationError(
|
||||
"version_label can only be used when merging one source document.",
|
||||
)
|
||||
if attrs["root_document_id"] not in documents:
|
||||
raise serializers.ValidationError(
|
||||
"root_document_id must be one of the selected documents.",
|
||||
)
|
||||
|
||||
selected_documents = Document.objects.filter(id__in=documents)
|
||||
if selected_documents.filter(root_document__isnull=False).exists():
|
||||
raise serializers.ValidationError(
|
||||
"Only top-level documents can be merged as versions.",
|
||||
)
|
||||
|
||||
source_document_ids = set(documents) - {attrs["root_document_id"]}
|
||||
if Document.global_objects.filter(
|
||||
root_document_id__in=source_document_ids,
|
||||
).exists():
|
||||
raise serializers.ValidationError(
|
||||
"Documents with existing versions cannot be merged into another document.",
|
||||
)
|
||||
return attrs
|
||||
|
||||
|
||||
class PdfEditOperationSerializer(serializers.Serializer[dict[str, int]]):
|
||||
page = serializers.IntegerField(min_value=1)
|
||||
rotate = serializers.IntegerField(required=False)
|
||||
doc = serializers.IntegerField(required=False, min_value=0)
|
||||
|
||||
def validate_rotate(self, value: int) -> int:
|
||||
return _validate_rotation_degrees(value, field="rotate")
|
||||
|
||||
|
||||
class EditPdfDocumentsSerializer(DocumentListSerializer, SourceModeValidationMixin):
|
||||
operations = serializers.ListField(
|
||||
child=PdfEditOperationSerializer(),
|
||||
required=True,
|
||||
allow_empty=False,
|
||||
)
|
||||
delete_original = serializers.BooleanField(required=False, default=False)
|
||||
update_document = serializers.BooleanField(required=False, default=False)
|
||||
include_metadata = serializers.BooleanField(required=False, default=True)
|
||||
source_mode = serializers.CharField(
|
||||
required=False,
|
||||
default=bulk_edit.SourceModeChoices.LATEST_VERSION,
|
||||
)
|
||||
from_webui = serializers.BooleanField(required=False, default=False)
|
||||
|
||||
def validate(self, attrs):
|
||||
documents = attrs["documents"]
|
||||
if len(documents) > 1:
|
||||
raise serializers.ValidationError(
|
||||
"Edit PDF method only supports one document",
|
||||
)
|
||||
|
||||
operations = attrs["operations"]
|
||||
|
||||
if any(op.get("doc", 0) >= len(operations) for op in operations):
|
||||
raise serializers.ValidationError("doc index is out of bounds")
|
||||
|
||||
if attrs["update_document"]:
|
||||
max_idx = max(op.get("doc", 0) for op in operations)
|
||||
if max_idx > 0:
|
||||
raise serializers.ValidationError(
|
||||
"update_document only allowed with a single output document",
|
||||
)
|
||||
|
||||
doc = Document.objects.get(id=documents[0])
|
||||
if doc.page_count:
|
||||
for op in operations:
|
||||
if op["page"] > doc.page_count:
|
||||
raise serializers.ValidationError(
|
||||
f"Page {op['page']} is out of bounds for document with {doc.page_count} pages.",
|
||||
)
|
||||
return attrs
|
||||
|
||||
|
||||
class RemovePasswordDocumentsSerializer(
|
||||
DocumentListSerializer,
|
||||
SourceModeValidationMixin,
|
||||
):
|
||||
password = serializers.CharField(required=True)
|
||||
update_document = serializers.BooleanField(required=False, default=False)
|
||||
delete_original = serializers.BooleanField(required=False, default=False)
|
||||
include_metadata = serializers.BooleanField(required=False, default=True)
|
||||
source_mode = serializers.CharField(
|
||||
required=False,
|
||||
default=bulk_edit.SourceModeChoices.LATEST_VERSION,
|
||||
)
|
||||
from_webui = serializers.BooleanField(required=False, default=False)
|
||||
|
||||
|
||||
class DeleteDocumentsSerializer(DocumentSelectionSerializer):
|
||||
pass
|
||||
|
||||
|
||||
class ReprocessDocumentsSerializer(DocumentSelectionSerializer):
|
||||
remote_ocr = serializers.BooleanField(required=False, default=False)
|
||||
|
||||
|
||||
class BulkEditSerializer(
|
||||
SerializerWithPerms,
|
||||
DocumentSelectionSerializer,
|
||||
SetPermissionsMixin,
|
||||
SourceModeValidationMixin,
|
||||
):
|
||||
# TODO: remove this and related backwards compatibility code when API v9 is dropped
|
||||
# split, delete_pages can be removed entirely
|
||||
MOVED_DOCUMENT_ACTION_ENDPOINTS = {
|
||||
"delete": "/api/documents/delete/",
|
||||
"reprocess": "/api/documents/reprocess/",
|
||||
"rotate": "/api/documents/rotate/",
|
||||
"merge": "/api/documents/merge/",
|
||||
"edit_pdf": "/api/documents/edit_pdf/",
|
||||
"remove_password": "/api/documents/remove_password/",
|
||||
"split": "/api/documents/edit_pdf/",
|
||||
"delete_pages": "/api/documents/edit_pdf/",
|
||||
}
|
||||
LEGACY_DOCUMENT_ACTION_METHODS = tuple(MOVED_DOCUMENT_ACTION_ENDPOINTS.keys())
|
||||
|
||||
method = serializers.ChoiceField(
|
||||
choices=[
|
||||
"set_correspondent",
|
||||
"set_document_type",
|
||||
"set_storage_path",
|
||||
"add_tag",
|
||||
"remove_tag",
|
||||
"modify_tags",
|
||||
"modify_custom_fields",
|
||||
"set_permissions",
|
||||
*LEGACY_DOCUMENT_ACTION_METHODS,
|
||||
],
|
||||
label="Method",
|
||||
write_only=True,
|
||||
)
|
||||
|
||||
parameters = serializers.DictField(allow_empty=True, default={}, write_only=True)
|
||||
from_webui = serializers.BooleanField(required=False, default=False)
|
||||
|
||||
def _validate_tag_id_list(self, tags, name="tags") -> None:
|
||||
if not isinstance(tags, list):
|
||||
raise serializers.ValidationError(f"{name} must be a list")
|
||||
if not all(isinstance(i, int) for i in tags):
|
||||
raise serializers.ValidationError(f"{name} must be a list of integers")
|
||||
count = Tag.objects.filter(id__in=tags).count()
|
||||
if not count == len(tags):
|
||||
raise serializers.ValidationError(
|
||||
f"Some tags in {name} don't exist or were specified twice.",
|
||||
)
|
||||
|
||||
def _validate_custom_field_id_list_or_dict(
|
||||
self,
|
||||
custom_fields,
|
||||
name="custom_fields",
|
||||
) -> None:
|
||||
ids = custom_fields
|
||||
if isinstance(custom_fields, dict):
|
||||
try:
|
||||
ids = [int(i[0]) for i in custom_fields.items()]
|
||||
except Exception as e:
|
||||
logger.exception(f"Error validating custom fields: {e}")
|
||||
raise serializers.ValidationError(
|
||||
f"{name} must be a list of integers or a dict of id:value pairs, see the log for details",
|
||||
)
|
||||
elif not isinstance(custom_fields, list) or not all(
|
||||
isinstance(i, int) for i in ids
|
||||
):
|
||||
raise serializers.ValidationError(
|
||||
f"{name} must be a list of integers or a dict of id:value pairs",
|
||||
)
|
||||
count = CustomField.objects.filter(id__in=ids).count()
|
||||
if not count == len(ids):
|
||||
raise serializers.ValidationError(
|
||||
f"Some custom fields in {name} don't exist or were specified twice.",
|
||||
)
|
||||
|
||||
def _validate_custom_field_values(self, custom_fields, name):
|
||||
if not isinstance(custom_fields, dict):
|
||||
return custom_fields
|
||||
|
||||
validated = {}
|
||||
errors = {}
|
||||
for raw_field_id, value in custom_fields.items():
|
||||
field_id = int(raw_field_id)
|
||||
validator = CustomFieldInstanceSerializer(
|
||||
data={"field": field_id, "value": value},
|
||||
context=self.context,
|
||||
)
|
||||
if validator.is_valid():
|
||||
validated[field_id] = validator.validated_data["value"]
|
||||
else:
|
||||
errors[str(field_id)] = validator.errors
|
||||
|
||||
if errors:
|
||||
raise serializers.ValidationError({name: errors})
|
||||
|
||||
return validated
|
||||
|
||||
def validate_method(self, method):
|
||||
if method == "set_correspondent":
|
||||
return bulk_edit.set_correspondent
|
||||
elif method == "set_document_type":
|
||||
return bulk_edit.set_document_type
|
||||
elif method == "set_storage_path":
|
||||
return bulk_edit.set_storage_path
|
||||
elif method == "add_tag":
|
||||
return bulk_edit.add_tag
|
||||
elif method == "remove_tag":
|
||||
return bulk_edit.remove_tag
|
||||
elif method == "modify_tags":
|
||||
return bulk_edit.modify_tags
|
||||
elif method == "modify_custom_fields":
|
||||
return bulk_edit.modify_custom_fields
|
||||
elif method == "delete":
|
||||
return bulk_edit.delete
|
||||
elif method == "redo_ocr" or method == "reprocess":
|
||||
return bulk_edit.reprocess
|
||||
elif method == "set_permissions":
|
||||
return bulk_edit.set_permissions
|
||||
elif method == "rotate":
|
||||
return bulk_edit.rotate
|
||||
elif method == "merge":
|
||||
return bulk_edit.merge
|
||||
elif method == "split":
|
||||
return bulk_edit.split
|
||||
elif method == "delete_pages":
|
||||
return bulk_edit.delete_pages
|
||||
elif method == "edit_pdf":
|
||||
return bulk_edit.edit_pdf
|
||||
elif method == "remove_password":
|
||||
return bulk_edit.remove_password
|
||||
else:
|
||||
raise serializers.ValidationError("Unsupported method.")
|
||||
|
||||
def _validate_parameters_tags(self, parameters) -> None:
|
||||
if "tag" in parameters:
|
||||
tag_id = parameters["tag"]
|
||||
try:
|
||||
Tag.objects.get(id=tag_id)
|
||||
except Tag.DoesNotExist:
|
||||
raise serializers.ValidationError("Tag does not exist")
|
||||
else:
|
||||
raise serializers.ValidationError("tag not specified")
|
||||
|
||||
def _validate_parameters_document_type(self, parameters) -> None:
|
||||
if "document_type" in parameters:
|
||||
document_type_id = parameters["document_type"]
|
||||
if document_type_id is None:
|
||||
# None is ok
|
||||
return
|
||||
try:
|
||||
DocumentType.objects.get(id=document_type_id)
|
||||
except DocumentType.DoesNotExist:
|
||||
raise serializers.ValidationError("Document type does not exist")
|
||||
else:
|
||||
raise serializers.ValidationError("document_type not specified")
|
||||
|
||||
def _validate_parameters_correspondent(self, parameters) -> None:
|
||||
if "correspondent" in parameters:
|
||||
correspondent_id = parameters["correspondent"]
|
||||
if correspondent_id is None:
|
||||
return
|
||||
try:
|
||||
Correspondent.objects.get(id=correspondent_id)
|
||||
except Correspondent.DoesNotExist:
|
||||
raise serializers.ValidationError("Correspondent does not exist")
|
||||
else:
|
||||
raise serializers.ValidationError("correspondent not specified")
|
||||
|
||||
def _validate_storage_path(self, parameters) -> None:
|
||||
if "storage_path" in parameters:
|
||||
storage_path_id = parameters["storage_path"]
|
||||
if storage_path_id is None:
|
||||
return
|
||||
try:
|
||||
StoragePath.objects.get(id=storage_path_id)
|
||||
except StoragePath.DoesNotExist:
|
||||
raise serializers.ValidationError(
|
||||
"Storage path does not exist",
|
||||
)
|
||||
else:
|
||||
raise serializers.ValidationError("storage path not specified")
|
||||
|
||||
def _validate_parameters_modify_tags(self, parameters) -> None:
|
||||
if "add_tags" in parameters:
|
||||
self._validate_tag_id_list(parameters["add_tags"], "add_tags")
|
||||
else:
|
||||
raise serializers.ValidationError("add_tags not specified")
|
||||
|
||||
if "remove_tags" in parameters:
|
||||
self._validate_tag_id_list(parameters["remove_tags"], "remove_tags")
|
||||
else:
|
||||
raise serializers.ValidationError("remove_tags not specified")
|
||||
|
||||
def _validate_parameters_modify_custom_fields(self, parameters) -> None:
|
||||
if "add_custom_fields" in parameters:
|
||||
self._validate_custom_field_id_list_or_dict(
|
||||
parameters["add_custom_fields"],
|
||||
"add_custom_fields",
|
||||
)
|
||||
parameters["add_custom_fields"] = self._validate_custom_field_values(
|
||||
parameters["add_custom_fields"],
|
||||
"add_custom_fields",
|
||||
)
|
||||
else:
|
||||
raise serializers.ValidationError("add_custom_fields not specified")
|
||||
|
||||
if "remove_custom_fields" in parameters:
|
||||
self._validate_custom_field_id_list_or_dict(
|
||||
parameters["remove_custom_fields"],
|
||||
"remove_custom_fields",
|
||||
)
|
||||
else:
|
||||
raise serializers.ValidationError("remove_custom_fields not specified")
|
||||
|
||||
def _validate_owner(self, owner) -> User:
|
||||
owner_field = serializers.PrimaryKeyRelatedField(queryset=User.objects.all())
|
||||
try:
|
||||
return owner_field.run_validation(owner)
|
||||
except serializers.ValidationError as e:
|
||||
raise serializers.ValidationError(
|
||||
"Specified owner cannot be found",
|
||||
) from e
|
||||
|
||||
def _validate_parameters_set_permissions(self, parameters) -> None:
|
||||
if "set_permissions" not in parameters:
|
||||
raise serializers.ValidationError("set_permissions not specified")
|
||||
set_permissions = parameters["set_permissions"]
|
||||
if set_permissions is not None:
|
||||
set_permissions = SetPermissionsSerializer().run_validation(
|
||||
set_permissions,
|
||||
)
|
||||
parameters["set_permissions"] = self.validate_set_permissions(
|
||||
set_permissions,
|
||||
)
|
||||
if "owner" in parameters and parameters["owner"] is not None:
|
||||
parameters["owner"] = self._validate_owner(parameters["owner"]).pk
|
||||
if "merge" not in parameters:
|
||||
parameters["merge"] = False
|
||||
|
||||
def _validate_parameters_rotate(self, parameters) -> None:
|
||||
if "degrees" not in parameters:
|
||||
raise serializers.ValidationError("invalid rotation degrees")
|
||||
try:
|
||||
degrees = serializers.IntegerField().run_validation(parameters["degrees"])
|
||||
except serializers.ValidationError as e:
|
||||
raise serializers.ValidationError("invalid rotation degrees") from e
|
||||
parameters["degrees"] = _validate_rotation_degrees(degrees)
|
||||
|
||||
def _validate_source_mode(self, parameters) -> None:
|
||||
source_mode = parameters.get(
|
||||
"source_mode",
|
||||
bulk_edit.SourceModeChoices.LATEST_VERSION,
|
||||
)
|
||||
parameters["source_mode"] = self.validate_source_mode(source_mode)
|
||||
|
||||
def _validate_parameters_split(self, parameters, document_id) -> None:
|
||||
if "pages" not in parameters:
|
||||
raise serializers.ValidationError("pages not specified")
|
||||
if not isinstance(parameters["pages"], str):
|
||||
raise serializers.ValidationError("invalid pages specified")
|
||||
page_count = Document.objects.get(id=document_id).page_count
|
||||
if not page_count:
|
||||
raise serializers.ValidationError("document page count is unknown")
|
||||
pages = []
|
||||
for group in parameters["pages"].split(","):
|
||||
start, is_range, end = group.partition("-")
|
||||
try:
|
||||
first = int(start)
|
||||
last = int(end) if is_range else first
|
||||
except ValueError as e:
|
||||
raise serializers.ValidationError("invalid pages specified") from e
|
||||
# Bound the range before building it, a huge one would exhaust memory
|
||||
if not 1 <= first <= last <= page_count:
|
||||
raise serializers.ValidationError("invalid pages specified")
|
||||
pages.append(list(range(first, last + 1)))
|
||||
parameters["pages"] = pages
|
||||
|
||||
if "delete_originals" in parameters:
|
||||
if not isinstance(parameters["delete_originals"], bool):
|
||||
raise serializers.ValidationError("delete_originals must be a boolean")
|
||||
else:
|
||||
parameters["delete_originals"] = False
|
||||
|
||||
def _validate_parameters_delete_pages(self, parameters) -> None:
|
||||
if "pages" not in parameters:
|
||||
raise serializers.ValidationError("pages not specified")
|
||||
if not isinstance(parameters["pages"], list):
|
||||
raise serializers.ValidationError("pages must be a list")
|
||||
if not all(isinstance(i, int) for i in parameters["pages"]):
|
||||
raise serializers.ValidationError("pages must be a list of integers")
|
||||
|
||||
def _validate_parameters_merge(self, parameters) -> None:
|
||||
if "delete_originals" in parameters:
|
||||
if not isinstance(parameters["delete_originals"], bool):
|
||||
raise serializers.ValidationError("delete_originals must be a boolean")
|
||||
else:
|
||||
parameters["delete_originals"] = False
|
||||
if "archive_fallback" in parameters:
|
||||
if not isinstance(parameters["archive_fallback"], bool):
|
||||
raise serializers.ValidationError("archive_fallback must be a boolean")
|
||||
else:
|
||||
parameters["archive_fallback"] = False
|
||||
|
||||
def _validate_parameters_edit_pdf(self, parameters, document_id) -> None:
|
||||
if "operations" not in parameters:
|
||||
raise serializers.ValidationError("operations not specified")
|
||||
operations_field = serializers.ListField(
|
||||
child=PdfEditOperationSerializer(),
|
||||
allow_empty=False,
|
||||
)
|
||||
try:
|
||||
operations = operations_field.run_validation(parameters["operations"])
|
||||
except serializers.ValidationError as e:
|
||||
# Key the errors under "operations" so they match what the
|
||||
# dedicated edit_pdf endpoint returns
|
||||
raise serializers.ValidationError({"operations": e.detail}) from e
|
||||
parameters["operations"] = operations
|
||||
|
||||
if "update_document" in parameters:
|
||||
if not isinstance(parameters["update_document"], bool):
|
||||
raise serializers.ValidationError("update_document must be a boolean")
|
||||
else:
|
||||
parameters["update_document"] = False
|
||||
if "include_metadata" in parameters:
|
||||
if not isinstance(parameters["include_metadata"], bool):
|
||||
raise serializers.ValidationError("include_metadata must be a boolean")
|
||||
else:
|
||||
parameters["include_metadata"] = True
|
||||
|
||||
if any(op.get("doc", 0) >= len(operations) for op in operations):
|
||||
raise serializers.ValidationError("doc index is out of bounds")
|
||||
|
||||
if parameters["update_document"]:
|
||||
max_idx = max(op.get("doc", 0) for op in operations)
|
||||
if max_idx > 0:
|
||||
raise serializers.ValidationError(
|
||||
"update_document only allowed with a single output document",
|
||||
)
|
||||
|
||||
doc = Document.objects.get(id=document_id)
|
||||
# doc existence is already validated
|
||||
if doc.page_count:
|
||||
for op in operations:
|
||||
if op["page"] > doc.page_count:
|
||||
raise serializers.ValidationError(
|
||||
f"Page {op['page']} is out of bounds for document with {doc.page_count} pages.",
|
||||
)
|
||||
|
||||
def _validate_parameters_reprocess(self, parameters) -> None:
|
||||
if "remote_ocr" in parameters:
|
||||
if not isinstance(parameters["remote_ocr"], bool):
|
||||
raise serializers.ValidationError("remote_ocr must be a boolean")
|
||||
else:
|
||||
parameters["remote_ocr"] = False
|
||||
|
||||
def validate_parameters_remove_password(self, parameters):
|
||||
if "password" not in parameters:
|
||||
raise serializers.ValidationError("password not specified")
|
||||
if not isinstance(parameters["password"], str):
|
||||
raise serializers.ValidationError("password must be a string")
|
||||
|
||||
def validate(self, attrs):
|
||||
attrs = super().validate(attrs)
|
||||
|
||||
if attrs.get("all", False) and attrs["method"] in [
|
||||
bulk_edit.merge,
|
||||
bulk_edit.split,
|
||||
bulk_edit.delete_pages,
|
||||
bulk_edit.edit_pdf,
|
||||
bulk_edit.remove_password,
|
||||
]:
|
||||
raise serializers.ValidationError(
|
||||
"This method does not support all=true.",
|
||||
)
|
||||
|
||||
method = attrs["method"]
|
||||
parameters = attrs["parameters"]
|
||||
|
||||
if "source_mode" in parameters:
|
||||
self._validate_source_mode(parameters)
|
||||
|
||||
if method == bulk_edit.set_correspondent:
|
||||
self._validate_parameters_correspondent(parameters)
|
||||
elif method == bulk_edit.set_document_type:
|
||||
self._validate_parameters_document_type(parameters)
|
||||
elif method == bulk_edit.add_tag or method == bulk_edit.remove_tag:
|
||||
self._validate_parameters_tags(parameters)
|
||||
elif method == bulk_edit.modify_tags:
|
||||
self._validate_parameters_modify_tags(parameters)
|
||||
elif method == bulk_edit.set_storage_path:
|
||||
self._validate_storage_path(parameters)
|
||||
elif method == bulk_edit.modify_custom_fields:
|
||||
self._validate_parameters_modify_custom_fields(parameters)
|
||||
elif method == bulk_edit.set_permissions:
|
||||
self._validate_parameters_set_permissions(parameters)
|
||||
elif method == bulk_edit.rotate:
|
||||
self._validate_parameters_rotate(parameters)
|
||||
elif method == bulk_edit.split:
|
||||
if len(attrs["documents"]) > 1:
|
||||
raise serializers.ValidationError(
|
||||
"Split method only supports one document",
|
||||
)
|
||||
self._validate_parameters_split(parameters, attrs["documents"][0])
|
||||
elif method == bulk_edit.delete_pages:
|
||||
if len(attrs["documents"]) > 1:
|
||||
raise serializers.ValidationError(
|
||||
"Delete pages method only supports one document",
|
||||
)
|
||||
self._validate_parameters_delete_pages(parameters)
|
||||
elif method == bulk_edit.merge:
|
||||
self._validate_parameters_merge(parameters)
|
||||
elif method == bulk_edit.edit_pdf:
|
||||
if len(attrs["documents"]) > 1:
|
||||
raise serializers.ValidationError(
|
||||
"Edit PDF method only supports one document",
|
||||
)
|
||||
self._validate_parameters_edit_pdf(parameters, attrs["documents"][0])
|
||||
elif method == bulk_edit.remove_password:
|
||||
self.validate_parameters_remove_password(parameters)
|
||||
elif method == bulk_edit.reprocess:
|
||||
self._validate_parameters_reprocess(parameters)
|
||||
|
||||
return attrs
|
||||
|
||||
|
||||
class BulkDownloadSerializer(DocumentSelectionSerializer):
|
||||
content = serializers.ChoiceField(
|
||||
choices=["archive", "originals", "both"],
|
||||
default="archive",
|
||||
)
|
||||
|
||||
compression = serializers.ChoiceField(
|
||||
choices=["none", "deflated", "bzip2", "lzma"],
|
||||
default="none",
|
||||
)
|
||||
|
||||
follow_formatting = serializers.BooleanField(
|
||||
default=False,
|
||||
)
|
||||
|
||||
def validate_compression(self, compression):
|
||||
import zipfile
|
||||
|
||||
return {
|
||||
"none": zipfile.ZIP_STORED,
|
||||
"deflated": zipfile.ZIP_DEFLATED,
|
||||
"bzip2": zipfile.ZIP_BZIP2,
|
||||
"lzma": zipfile.ZIP_LZMA,
|
||||
}[compression]
|
||||
|
||||
|
||||
class BulkEditObjectsSerializer(SerializerWithPerms, SetPermissionsMixin):
|
||||
objects = serializers.ListField(
|
||||
required=False,
|
||||
allow_empty=True,
|
||||
label="Objects",
|
||||
write_only=True,
|
||||
child=serializers.IntegerField(),
|
||||
)
|
||||
|
||||
all = serializers.BooleanField(
|
||||
default=False,
|
||||
required=False,
|
||||
write_only=True,
|
||||
)
|
||||
|
||||
filters = serializers.DictField(
|
||||
required=False,
|
||||
allow_empty=True,
|
||||
write_only=True,
|
||||
)
|
||||
|
||||
object_type = serializers.ChoiceField(
|
||||
choices=[
|
||||
"tags",
|
||||
"correspondents",
|
||||
"document_types",
|
||||
"storage_paths",
|
||||
],
|
||||
label="Object Type",
|
||||
write_only=True,
|
||||
)
|
||||
|
||||
operation = serializers.ChoiceField(
|
||||
choices=[
|
||||
"set_permissions",
|
||||
"delete",
|
||||
],
|
||||
label="Operation",
|
||||
required=True,
|
||||
write_only=True,
|
||||
)
|
||||
|
||||
owner = serializers.PrimaryKeyRelatedField(
|
||||
queryset=User.objects.all(),
|
||||
required=False,
|
||||
allow_null=True,
|
||||
)
|
||||
|
||||
permissions = SetPermissionsSerializer(
|
||||
label="Set permissions",
|
||||
required=False,
|
||||
write_only=True,
|
||||
)
|
||||
|
||||
merge = serializers.BooleanField(
|
||||
default=False,
|
||||
write_only=True,
|
||||
required=False,
|
||||
)
|
||||
|
||||
def get_object_class(self, object_type):
|
||||
object_class = None
|
||||
if object_type == "tags":
|
||||
object_class = Tag
|
||||
elif object_type == "correspondents":
|
||||
object_class = Correspondent
|
||||
elif object_type == "document_types":
|
||||
object_class = DocumentType
|
||||
elif object_type == "storage_paths":
|
||||
object_class = StoragePath
|
||||
return object_class
|
||||
|
||||
def _validate_objects(self, objects, object_type):
|
||||
if not isinstance(objects, list):
|
||||
raise serializers.ValidationError("objects must be a list")
|
||||
if not all(isinstance(i, int) for i in objects):
|
||||
raise serializers.ValidationError("objects must be a list of integers")
|
||||
object_class = self.get_object_class(object_type)
|
||||
count = object_class.objects.filter(id__in=objects).count()
|
||||
if not count == len(objects):
|
||||
raise serializers.ValidationError(
|
||||
"Some ids in objects don't exist or were specified twice.",
|
||||
)
|
||||
return objects
|
||||
|
||||
def _validate_permissions(self, permissions) -> dict:
|
||||
return self.validate_set_permissions(
|
||||
permissions,
|
||||
)
|
||||
|
||||
def validate(self, attrs):
|
||||
object_type = attrs["object_type"]
|
||||
objects = attrs.get("objects")
|
||||
apply_to_all = attrs.get("all", False)
|
||||
operation = attrs.get("operation")
|
||||
|
||||
if apply_to_all:
|
||||
attrs.setdefault("objects", [])
|
||||
else:
|
||||
if objects is None:
|
||||
raise serializers.ValidationError(
|
||||
"objects is required unless all is true.",
|
||||
)
|
||||
if len(objects) == 0:
|
||||
raise serializers.ValidationError("objects must not be empty")
|
||||
self._validate_objects(objects, object_type)
|
||||
|
||||
if operation == "set_permissions":
|
||||
permissions = attrs.get("permissions")
|
||||
if permissions is not None:
|
||||
if not permissions:
|
||||
raise serializers.ValidationError(
|
||||
"permissions must not be empty",
|
||||
)
|
||||
attrs["permissions"] = self._validate_permissions(permissions)
|
||||
|
||||
return attrs
|
||||
@@ -0,0 +1,474 @@
|
||||
from __future__ import annotations
|
||||
|
||||
import logging
|
||||
from typing import TYPE_CHECKING
|
||||
from typing import Any
|
||||
from typing import TypedDict
|
||||
|
||||
from django.conf import settings
|
||||
from django.contrib.auth.models import User
|
||||
from django.db.models import Q
|
||||
from django.utils.dateparse import parse_datetime
|
||||
from django.utils.timezone import get_current_timezone
|
||||
from django.utils.timezone import is_naive
|
||||
from django.utils.timezone import make_aware
|
||||
from drf_spectacular.utils import extend_schema_field
|
||||
from drf_spectacular.utils import extend_schema_serializer
|
||||
from drf_writable_nested.serializers import NestedUpdateMixin
|
||||
from rest_framework import serializers
|
||||
from rest_framework.fields import SerializerMethodField
|
||||
|
||||
if settings.AUDIT_LOG_ENABLED:
|
||||
from auditlog.context import set_actor
|
||||
|
||||
|
||||
from documents import bulk_edit
|
||||
from documents.models import CustomField
|
||||
from documents.models import CustomFieldInstance
|
||||
from documents.models import Document
|
||||
from documents.models import Tag
|
||||
from documents.permissions import permitted_document_ids
|
||||
from documents.versioning import has_prefetched_effective_content
|
||||
from documents.versioning import sort_versions_newest_first
|
||||
|
||||
from .base import DocumentUpdateFieldsModelSerializer
|
||||
from .base import NotesSerializer
|
||||
from .base import OwnedObjectListSerializer
|
||||
from .base import OwnedObjectSerializer
|
||||
from .metadata import CorrespondentField
|
||||
from .metadata import CustomFieldInstanceSerializer
|
||||
from .metadata import DocumentTypeField
|
||||
from .metadata import StoragePathField
|
||||
from .metadata import TagsField
|
||||
from .upload import PostDocumentSerializer
|
||||
|
||||
if TYPE_CHECKING:
|
||||
from datetime import datetime
|
||||
|
||||
from django.db.models.query import QuerySet
|
||||
from rest_framework.relations import ManyRelatedField
|
||||
from rest_framework.relations import RelatedField
|
||||
|
||||
|
||||
logger = logging.getLogger("paperless.serializers")
|
||||
|
||||
|
||||
def _get_viewable_duplicates(
|
||||
document: Document,
|
||||
user: User | None,
|
||||
) -> QuerySet[Document]:
|
||||
checksums = {document.checksum}
|
||||
if document.archive_checksum:
|
||||
checksums.add(document.archive_checksum)
|
||||
duplicates = Document.global_objects.filter(
|
||||
Q(checksum__in=checksums) | Q(archive_checksum__in=checksums),
|
||||
).exclude(pk=document.pk)
|
||||
duplicates = duplicates.filter(root_document__isnull=True)
|
||||
duplicates = duplicates.order_by("-created")
|
||||
allowed_ids = permitted_document_ids(user, include_deleted=True)
|
||||
return duplicates.filter(id__in=allowed_ids)
|
||||
|
||||
|
||||
class DuplicateDocumentSummarySerializer(serializers.Serializer[dict[str, Any]]):
|
||||
id = serializers.IntegerField()
|
||||
title = serializers.CharField()
|
||||
deleted_at = serializers.DateTimeField(allow_null=True)
|
||||
|
||||
|
||||
class _DocumentVersionInfo(TypedDict):
|
||||
id: int
|
||||
added: datetime
|
||||
version_label: str | None
|
||||
checksum: str | None
|
||||
is_root: bool
|
||||
|
||||
|
||||
class DocumentVersionInfoSerializer(serializers.Serializer[_DocumentVersionInfo]):
|
||||
id = serializers.IntegerField()
|
||||
added = serializers.DateTimeField()
|
||||
version_label = serializers.CharField(required=False, allow_null=True)
|
||||
checksum = serializers.CharField(required=False, allow_null=True)
|
||||
is_root = serializers.BooleanField()
|
||||
|
||||
|
||||
@extend_schema_serializer(
|
||||
deprecate_fields=["created_date"],
|
||||
)
|
||||
class DocumentSerializer(
|
||||
OwnedObjectSerializer,
|
||||
NestedUpdateMixin,
|
||||
DocumentUpdateFieldsModelSerializer,
|
||||
):
|
||||
correspondent = CorrespondentField(allow_null=True)
|
||||
tags = TagsField(many=True)
|
||||
document_type = DocumentTypeField(allow_null=True)
|
||||
storage_path = StoragePathField(allow_null=True)
|
||||
|
||||
original_file_name = SerializerMethodField()
|
||||
archived_file_name = SerializerMethodField()
|
||||
created_date = serializers.DateField(required=False)
|
||||
page_count = SerializerMethodField()
|
||||
duplicate_documents = SerializerMethodField()
|
||||
|
||||
notes = NotesSerializer(many=True, required=False, read_only=True)
|
||||
root_document: RelatedField[Document, Document, Any] | ManyRelatedField = (
|
||||
serializers.PrimaryKeyRelatedField(read_only=True)
|
||||
)
|
||||
versions = SerializerMethodField()
|
||||
|
||||
custom_fields = CustomFieldInstanceSerializer(
|
||||
many=True,
|
||||
allow_null=False,
|
||||
required=False,
|
||||
)
|
||||
|
||||
owner = serializers.PrimaryKeyRelatedField(
|
||||
queryset=User.objects.all(),
|
||||
required=False,
|
||||
allow_null=True,
|
||||
)
|
||||
|
||||
remove_inbox_tags = serializers.BooleanField(
|
||||
default=False,
|
||||
write_only=True,
|
||||
allow_null=True,
|
||||
required=False,
|
||||
)
|
||||
|
||||
def get_page_count(self, obj) -> int | None:
|
||||
return obj.page_count
|
||||
|
||||
@extend_schema_field(DuplicateDocumentSummarySerializer(many=True))
|
||||
def get_duplicate_documents(self, obj):
|
||||
view = self.context.get("view")
|
||||
if view and getattr(view, "action", None) != "retrieve":
|
||||
return []
|
||||
request = self.context.get("request")
|
||||
user = request.user if request else None
|
||||
duplicates = _get_viewable_duplicates(obj, user)
|
||||
return list(duplicates.values("id", "title", "deleted_at"))
|
||||
|
||||
@extend_schema_field(DocumentVersionInfoSerializer(many=True))
|
||||
def get_versions(self, obj):
|
||||
root_doc = obj if obj.root_document_id is None else obj.root_document
|
||||
if root_doc is None:
|
||||
return []
|
||||
|
||||
prefetched_cache = getattr(obj, "_prefetched_objects_cache", None)
|
||||
prefetched_versions = (
|
||||
prefetched_cache.get("versions")
|
||||
if isinstance(prefetched_cache, dict)
|
||||
else None
|
||||
)
|
||||
|
||||
versions: list[Document]
|
||||
if prefetched_versions is not None:
|
||||
versions = [*prefetched_versions, root_doc]
|
||||
else:
|
||||
versions_qs = Document.objects.filter(root_document=root_doc).only(
|
||||
"id",
|
||||
"added",
|
||||
"checksum",
|
||||
"version_label",
|
||||
"root_document_id",
|
||||
"version_index",
|
||||
)
|
||||
versions = [*versions_qs, root_doc]
|
||||
|
||||
versions = sort_versions_newest_first(versions)
|
||||
|
||||
def build_info(doc: Document) -> _DocumentVersionInfo:
|
||||
return {
|
||||
"id": doc.id,
|
||||
"added": doc.added,
|
||||
"version_label": doc.version_label,
|
||||
"checksum": doc.checksum,
|
||||
"is_root": doc.id == root_doc.id,
|
||||
}
|
||||
|
||||
return [build_info(doc) for doc in versions]
|
||||
|
||||
def get_original_file_name(self, obj) -> str | None:
|
||||
return obj.original_filename
|
||||
|
||||
def get_archived_file_name(self, obj) -> str | None:
|
||||
if obj.has_archive_version:
|
||||
return obj.get_public_filename(archive=True)
|
||||
else:
|
||||
return None
|
||||
|
||||
def to_representation(self, instance):
|
||||
doc = super().to_representation(instance)
|
||||
if "content" in self.fields and has_prefetched_effective_content(instance):
|
||||
# Only resolve version-aware content when it's cheap: an SQL
|
||||
# annotation or a versions prefetch is already on the instance.
|
||||
# A caller that set up neither (e.g. TrashView, GlobalSearchView,
|
||||
# which build their own querysets) gets the document's own,
|
||||
# unresolved content instead of paying for an extra per-instance
|
||||
# query -- same as before effective_content resolution existed.
|
||||
doc["content"] = instance.get_effective_content() or ""
|
||||
if self.truncate_content and "content" in self.fields:
|
||||
doc["content"] = doc.get("content")[0:550]
|
||||
return doc
|
||||
|
||||
def to_internal_value(self, data):
|
||||
if (
|
||||
"created" in data
|
||||
and isinstance(data["created"], str)
|
||||
and ":" in data["created"]
|
||||
):
|
||||
# Handle old format of isoformat datetime string
|
||||
parsed = parse_datetime(data["created"])
|
||||
if parsed:
|
||||
if is_naive(parsed):
|
||||
parsed = make_aware(parsed, get_current_timezone())
|
||||
data["created"] = parsed.astimezone().date()
|
||||
return super().to_internal_value(data)
|
||||
|
||||
def validate(self, attrs):
|
||||
if (
|
||||
"archive_serial_number" in attrs
|
||||
and attrs["archive_serial_number"] is not None
|
||||
and len(str(attrs["archive_serial_number"])) > 0
|
||||
and Document.deleted_objects.filter(
|
||||
archive_serial_number=attrs["archive_serial_number"],
|
||||
).exists()
|
||||
):
|
||||
raise serializers.ValidationError(
|
||||
{
|
||||
"archive_serial_number": [
|
||||
"Document with this Archive Serial Number already exists in the trash.",
|
||||
],
|
||||
},
|
||||
)
|
||||
return super().validate(attrs)
|
||||
|
||||
def update(self, instance: Document, validated_data):
|
||||
if "created_date" in validated_data:
|
||||
if "created" not in validated_data:
|
||||
validated_data["created"] = validated_data["created_date"]
|
||||
logger.warning(
|
||||
"created_date is deprecated, use created instead",
|
||||
)
|
||||
validated_data.pop("created_date")
|
||||
if instance.custom_fields.count() > 0 and "custom_fields" in validated_data:
|
||||
incoming_custom_fields = [
|
||||
field["field"] for field in validated_data["custom_fields"]
|
||||
]
|
||||
for custom_field_instance in instance.custom_fields.filter(
|
||||
field__data_type=CustomField.FieldDataType.DOCUMENTLINK,
|
||||
):
|
||||
if (
|
||||
custom_field_instance.field not in incoming_custom_fields
|
||||
and custom_field_instance.value is not None
|
||||
):
|
||||
# Doc link field is being removed entirely
|
||||
for doc_id in custom_field_instance.value:
|
||||
bulk_edit.remove_doclink(
|
||||
instance,
|
||||
custom_field_instance.field,
|
||||
doc_id,
|
||||
)
|
||||
if "tags" in validated_data:
|
||||
# Respect tag hierarchy on updates:
|
||||
# - Adding a child adds its ancestors
|
||||
# - Removing a parent removes all its descendants
|
||||
prev_tags = set(instance.tags.all())
|
||||
requested_tags = set(validated_data["tags"])
|
||||
|
||||
# Tags newly added in this update and the ancestors they require
|
||||
added_tags = requested_tags - prev_tags
|
||||
required_by_add_tags = set(added_tags)
|
||||
for t in added_tags:
|
||||
required_by_add_tags.update(t.get_ancestors())
|
||||
|
||||
# Tags being removed in this update and all descendants, except
|
||||
# those required by a tag that is being added in this same update
|
||||
removed_tags = prev_tags - requested_tags
|
||||
blocked_tags = set(removed_tags)
|
||||
for t in removed_tags:
|
||||
blocked_tags.update(t.get_descendants())
|
||||
blocked_tags.difference_update(required_by_add_tags)
|
||||
|
||||
# Add all parent tags
|
||||
final_tags = set(requested_tags)
|
||||
for t in requested_tags:
|
||||
final_tags.update(t.get_ancestors())
|
||||
|
||||
# Drop removed parents and their descendants
|
||||
final_tags.difference_update(blocked_tags)
|
||||
|
||||
validated_data["tags"] = list(final_tags)
|
||||
if validated_data.get("remove_inbox_tags"):
|
||||
current_tag_ids = {t.pk for t in instance.tags.all()}
|
||||
tags = (
|
||||
validated_data["tags"]
|
||||
if "tags" in validated_data
|
||||
else list(instance.tags.all())
|
||||
)
|
||||
|
||||
# Tags newly added in this update, plus their ancestors, are kept
|
||||
keep_ids: set[int] = set()
|
||||
for tag in tags:
|
||||
if tag.pk not in current_tag_ids:
|
||||
keep_ids.add(tag.pk)
|
||||
keep_ids.update(int(pk) for pk in tag.get_ancestors_pks())
|
||||
|
||||
# Remove inbox tags and their descendants, except those being kept
|
||||
remove_ids: set[int] = set()
|
||||
for inbox_tag in (
|
||||
Tag.objects.filter(is_inbox_tag=True)
|
||||
.exclude(pk__in=keep_ids)
|
||||
.only("pk", "tn_descendants_pks")
|
||||
):
|
||||
remove_ids.add(inbox_tag.pk)
|
||||
remove_ids.update(int(pk) for pk in inbox_tag.get_descendants_pks())
|
||||
|
||||
validated_data["tags"] = [t for t in tags if t.pk not in remove_ids]
|
||||
|
||||
if settings.AUDIT_LOG_ENABLED:
|
||||
with set_actor(self.user):
|
||||
super().update(instance, validated_data)
|
||||
else:
|
||||
super().update(instance, validated_data)
|
||||
|
||||
# hard delete custom field instances that were soft deleted
|
||||
CustomFieldInstance.deleted_objects.filter(document=instance).delete()
|
||||
return instance
|
||||
|
||||
def __init__(self, *args, **kwargs) -> None:
|
||||
self.truncate_content = kwargs.pop("truncate_content", False)
|
||||
|
||||
# return full permissions if we're doing a PATCH or PUT
|
||||
context = kwargs.get("context")
|
||||
if context is not None and (
|
||||
context.get("request").method == "PATCH"
|
||||
or context.get("request").method == "PUT"
|
||||
):
|
||||
kwargs["full_perms"] = True
|
||||
|
||||
super().__init__(*args, **kwargs)
|
||||
|
||||
class Meta:
|
||||
model = Document
|
||||
fields = (
|
||||
"id",
|
||||
"correspondent",
|
||||
"document_type",
|
||||
"storage_path",
|
||||
"title",
|
||||
"content",
|
||||
"tags",
|
||||
"created",
|
||||
"created_date",
|
||||
"modified",
|
||||
"added",
|
||||
"deleted_at",
|
||||
"archive_serial_number",
|
||||
"original_file_name",
|
||||
"archived_file_name",
|
||||
"duplicate_documents",
|
||||
"owner",
|
||||
"permissions",
|
||||
"user_can_change",
|
||||
"is_shared_by_requester",
|
||||
"set_permissions",
|
||||
"notes",
|
||||
"custom_fields",
|
||||
"remove_inbox_tags",
|
||||
"page_count",
|
||||
"mime_type",
|
||||
"root_document",
|
||||
"versions",
|
||||
)
|
||||
read_only_fields = ("deleted_at",)
|
||||
list_serializer_class = OwnedObjectListSerializer
|
||||
|
||||
|
||||
class SearchResultListSerializer(serializers.ListSerializer[Document]):
|
||||
def to_representation(self, hits):
|
||||
document_ids = [hit["id"] for hit in hits]
|
||||
# Fetch all Document objects in the list in one SQL query.
|
||||
documents = self.child.fetch_documents(document_ids)
|
||||
self.child.context["documents"] = documents
|
||||
# Also check if they are shared with other users / groups.
|
||||
self.child.context["shared_object_pks"] = self.child.get_shared_object_pks(
|
||||
documents.values(),
|
||||
)
|
||||
|
||||
return super().to_representation(hits)
|
||||
|
||||
|
||||
class SearchResultSerializer(DocumentSerializer):
|
||||
@staticmethod
|
||||
def fetch_documents(ids):
|
||||
"""
|
||||
Return a dict that maps given document IDs to Document objects.
|
||||
"""
|
||||
return {
|
||||
document.id: document
|
||||
for document in Document.objects.select_related(
|
||||
"correspondent",
|
||||
"storage_path",
|
||||
"document_type",
|
||||
"owner",
|
||||
)
|
||||
.prefetch_related("tags", "custom_fields", "notes")
|
||||
.filter(id__in=ids)
|
||||
}
|
||||
|
||||
def to_representation(self, hit):
|
||||
# Again we first check if the parent has already fetched the documents.
|
||||
documents = self.context.get("documents")
|
||||
# Otherwise we fetch this document.
|
||||
if documents is None: # pragma: no cover
|
||||
# In practice we only serialize **lists** of SearchHit dicts.
|
||||
# Keeping this check for completeness but marking it no cover for now.
|
||||
documents = self.fetch_documents([hit["id"]])
|
||||
document = documents[hit["id"]]
|
||||
|
||||
highlights = hit.get("highlights", {})
|
||||
r = super().to_representation(document)
|
||||
r["__search_hit__"] = {
|
||||
"score": hit["score"],
|
||||
"highlights": highlights.get("content", ""),
|
||||
"note_highlights": highlights.get("notes") or None,
|
||||
"rank": hit["rank"],
|
||||
}
|
||||
|
||||
return r
|
||||
|
||||
class Meta(DocumentSerializer.Meta):
|
||||
list_serializer_class = SearchResultListSerializer
|
||||
|
||||
|
||||
class DocumentVersionSerializer(serializers.Serializer[dict[str, Any]]):
|
||||
document = serializers.FileField(
|
||||
label="Document",
|
||||
write_only=True,
|
||||
)
|
||||
version_label = serializers.CharField(
|
||||
label="Version label",
|
||||
required=False,
|
||||
allow_blank=True,
|
||||
allow_null=True,
|
||||
max_length=64,
|
||||
)
|
||||
|
||||
validate_document = PostDocumentSerializer().validate_document
|
||||
|
||||
|
||||
class DocumentVersionLabelSerializer(serializers.Serializer[dict[str, str | None]]):
|
||||
version_label = serializers.CharField(
|
||||
label="Version label",
|
||||
required=True,
|
||||
allow_blank=True,
|
||||
allow_null=True,
|
||||
max_length=64,
|
||||
)
|
||||
|
||||
def validate_version_label(self, value):
|
||||
if value is None:
|
||||
return None
|
||||
normalized = value.strip()
|
||||
return normalized or None
|
||||
@@ -0,0 +1,557 @@
|
||||
from __future__ import annotations
|
||||
|
||||
import logging
|
||||
import math
|
||||
import re
|
||||
from decimal import Decimal
|
||||
from typing import Any
|
||||
|
||||
from django.core.exceptions import ValidationError
|
||||
from django.core.validators import DecimalValidator
|
||||
from django.core.validators import MaxLengthValidator
|
||||
from django.core.validators import MaxValueValidator
|
||||
from django.core.validators import MinValueValidator
|
||||
from django.core.validators import RegexValidator
|
||||
from django.core.validators import integer_validator
|
||||
from django.db.models import Count
|
||||
from django.db.models.functions import Lower
|
||||
from django.utils.crypto import get_random_string
|
||||
from django.utils.translation import gettext as _
|
||||
from drf_spectacular.utils import extend_schema_field
|
||||
from rest_framework import serializers
|
||||
from rest_framework.exceptions import PermissionDenied
|
||||
from rest_framework.filters import OrderingFilter
|
||||
|
||||
from documents import bulk_edit
|
||||
from documents.models import Correspondent
|
||||
from documents.models import CustomField
|
||||
from documents.models import CustomFieldInstance
|
||||
from documents.models import Document
|
||||
from documents.models import DocumentType
|
||||
from documents.models import PaperlessTask
|
||||
from documents.models import StoragePath
|
||||
from documents.models import Tag
|
||||
from documents.permissions import get_document_count_filter_for_user
|
||||
from documents.permissions import permitted_document_ids
|
||||
from documents.permissions import restrict_queryset_to_visible
|
||||
from documents.templating.filepath import validate_filepath_template_and_render
|
||||
from documents.templating.utils import convert_format_str_to_template_format
|
||||
from documents.validators import uri_validator
|
||||
|
||||
from .base import MatchingModelSerializer
|
||||
from .base import OwnedObjectSerializer
|
||||
from .base import ReadWriteSerializerMethodField
|
||||
from .base import SerializerWithPerms
|
||||
|
||||
logger = logging.getLogger("paperless.serializers")
|
||||
|
||||
|
||||
class CorrespondentSerializer(MatchingModelSerializer, OwnedObjectSerializer):
|
||||
last_correspondence = serializers.DateField(read_only=True, required=False)
|
||||
|
||||
class Meta:
|
||||
model = Correspondent
|
||||
fields = (
|
||||
"id",
|
||||
"slug",
|
||||
"name",
|
||||
"match",
|
||||
"matching_algorithm",
|
||||
"is_insensitive",
|
||||
"document_count",
|
||||
"last_correspondence",
|
||||
"owner",
|
||||
"permissions",
|
||||
"user_can_change",
|
||||
"set_permissions",
|
||||
)
|
||||
|
||||
|
||||
class DocumentTypeSerializer(MatchingModelSerializer, OwnedObjectSerializer):
|
||||
class Meta:
|
||||
model = DocumentType
|
||||
fields = (
|
||||
"id",
|
||||
"slug",
|
||||
"name",
|
||||
"match",
|
||||
"matching_algorithm",
|
||||
"is_insensitive",
|
||||
"document_count",
|
||||
"owner",
|
||||
"permissions",
|
||||
"user_can_change",
|
||||
"set_permissions",
|
||||
)
|
||||
|
||||
|
||||
class DeprecatedColors:
|
||||
COLOURS = (
|
||||
(1, "#a6cee3"),
|
||||
(2, "#1f78b4"),
|
||||
(3, "#b2df8a"),
|
||||
(4, "#33a02c"),
|
||||
(5, "#fb9a99"),
|
||||
(6, "#e31a1c"),
|
||||
(7, "#fdbf6f"),
|
||||
(8, "#ff7f00"),
|
||||
(9, "#cab2d6"),
|
||||
(10, "#6a3d9a"),
|
||||
(11, "#b15928"),
|
||||
(12, "#000000"),
|
||||
(13, "#cccccc"),
|
||||
)
|
||||
|
||||
|
||||
@extend_schema_field(
|
||||
serializers.ChoiceField(
|
||||
choices=DeprecatedColors.COLOURS,
|
||||
),
|
||||
)
|
||||
class ColorField(serializers.Field):
|
||||
def to_internal_value(self, data):
|
||||
for id, color in DeprecatedColors.COLOURS:
|
||||
if id == data:
|
||||
return color
|
||||
raise serializers.ValidationError
|
||||
|
||||
def to_representation(self, value):
|
||||
for id, color in DeprecatedColors.COLOURS:
|
||||
if color == value:
|
||||
return id
|
||||
return 1
|
||||
|
||||
|
||||
class TagSerializer(MatchingModelSerializer, OwnedObjectSerializer):
|
||||
def get_text_color(self, obj) -> str:
|
||||
try:
|
||||
h = obj.color.lstrip("#")
|
||||
rgb = tuple(int(h[i : i + 2], 16) / 256 for i in (0, 2, 4))
|
||||
luminance = math.sqrt(
|
||||
0.299 * math.pow(rgb[0], 2)
|
||||
+ 0.587 * math.pow(rgb[1], 2)
|
||||
+ 0.114 * math.pow(rgb[2], 2),
|
||||
)
|
||||
return "#ffffff" if luminance < 0.53 else "#000000"
|
||||
except ValueError:
|
||||
return "#000000"
|
||||
|
||||
text_color = serializers.SerializerMethodField()
|
||||
|
||||
# map to treenode's tn_parent
|
||||
parent = serializers.PrimaryKeyRelatedField(
|
||||
queryset=Tag.objects.all(),
|
||||
allow_null=True,
|
||||
required=False,
|
||||
source="tn_parent",
|
||||
)
|
||||
|
||||
@extend_schema_field(
|
||||
field=serializers.ListSerializer(
|
||||
child=serializers.PrimaryKeyRelatedField(
|
||||
queryset=Tag.objects.all(),
|
||||
),
|
||||
),
|
||||
)
|
||||
def get_children(self, obj):
|
||||
children_map = self.context.get("children_map")
|
||||
if children_map is not None:
|
||||
children = children_map.get(obj.pk, [])
|
||||
else:
|
||||
filter_q = self.context.get("document_count_filter")
|
||||
request = self.context.get("request")
|
||||
if filter_q is None:
|
||||
user = getattr(request, "user", None) if request else None
|
||||
filter_q = get_document_count_filter_for_user(user)
|
||||
self.context["document_count_filter"] = filter_q
|
||||
|
||||
children = (
|
||||
obj.get_children_queryset()
|
||||
.select_related("owner")
|
||||
.annotate(document_count=Count("documents", filter=filter_q))
|
||||
)
|
||||
user = getattr(request, "user", None) if request else self.user
|
||||
children = restrict_queryset_to_visible(children, user, "view_tag")
|
||||
|
||||
view = self.context.get("view")
|
||||
ordering = (
|
||||
OrderingFilter().get_ordering(request, children, view)
|
||||
if request and view
|
||||
else None
|
||||
)
|
||||
ordering = ordering or (Lower("name"),)
|
||||
children = children.order_by(*ordering)
|
||||
|
||||
if not children:
|
||||
return []
|
||||
|
||||
serializer = TagSerializer(
|
||||
children,
|
||||
many=True,
|
||||
user=self.user,
|
||||
full_perms=self.full_perms,
|
||||
all_fields=self.all_fields,
|
||||
context=self.context,
|
||||
)
|
||||
return serializer.data
|
||||
|
||||
# children as nested Tag objects
|
||||
children = serializers.SerializerMethodField()
|
||||
|
||||
class Meta:
|
||||
model = Tag
|
||||
fields = (
|
||||
"id",
|
||||
"slug",
|
||||
"name",
|
||||
"color",
|
||||
"text_color",
|
||||
"match",
|
||||
"matching_algorithm",
|
||||
"is_insensitive",
|
||||
"is_inbox_tag",
|
||||
"document_count",
|
||||
"owner",
|
||||
"permissions",
|
||||
"user_can_change",
|
||||
"set_permissions",
|
||||
"parent",
|
||||
"children",
|
||||
)
|
||||
|
||||
def validate_color(self, color):
|
||||
regex = r"#[0-9a-fA-F]{6}"
|
||||
if not re.match(regex, color):
|
||||
raise serializers.ValidationError(_("Invalid color."))
|
||||
return color
|
||||
|
||||
def validate(self, attrs):
|
||||
# Validate when changing parent
|
||||
parent = attrs.get(
|
||||
"tn_parent",
|
||||
self.instance.get_parent() if self.instance else None,
|
||||
)
|
||||
|
||||
if self.instance:
|
||||
# Temporarily set parent on the instance if updating and use model clean()
|
||||
original_parent = self.instance.get_parent()
|
||||
try:
|
||||
# Temporarily set tn_parent in-memory to validate clean()
|
||||
self.instance.tn_parent = parent
|
||||
self.instance.clean()
|
||||
except ValidationError as e:
|
||||
logger.debug("Tag parent validation failed: %s", e)
|
||||
raise e
|
||||
finally:
|
||||
self.instance.tn_parent = original_parent
|
||||
else:
|
||||
# For new instances, create a transient Tag and validate
|
||||
temp = Tag(tn_parent=parent)
|
||||
try:
|
||||
temp.clean()
|
||||
except ValidationError as e:
|
||||
logger.debug("Tag parent validation failed: %s", e)
|
||||
raise e
|
||||
|
||||
return super().validate(attrs)
|
||||
|
||||
|
||||
class CorrespondentField(serializers.PrimaryKeyRelatedField[Correspondent]):
|
||||
def get_queryset(self):
|
||||
return Correspondent.objects.all()
|
||||
|
||||
|
||||
class TagsField(serializers.PrimaryKeyRelatedField[Tag]):
|
||||
def get_queryset(self):
|
||||
return Tag.objects.all()
|
||||
|
||||
|
||||
class DocumentTypeField(serializers.PrimaryKeyRelatedField[DocumentType]):
|
||||
def get_queryset(self):
|
||||
return DocumentType.objects.all()
|
||||
|
||||
|
||||
class StoragePathField(serializers.PrimaryKeyRelatedField[StoragePath]):
|
||||
def get_queryset(self):
|
||||
return StoragePath.objects.all()
|
||||
|
||||
|
||||
class CustomFieldSerializer(serializers.ModelSerializer[CustomField]):
|
||||
data_type = serializers.ChoiceField(
|
||||
choices=CustomField.FieldDataType,
|
||||
read_only=False,
|
||||
)
|
||||
|
||||
document_count = serializers.IntegerField(read_only=True)
|
||||
|
||||
class Meta:
|
||||
model = CustomField
|
||||
fields = [
|
||||
"id",
|
||||
"name",
|
||||
"data_type",
|
||||
"extra_data",
|
||||
"document_count",
|
||||
]
|
||||
|
||||
def validate(self, attrs):
|
||||
# TODO: remove pending https://github.com/encode/django-rest-framework/issues/7173
|
||||
name = attrs.get(
|
||||
"name",
|
||||
self.instance.name if hasattr(self.instance, "name") else None,
|
||||
)
|
||||
objects = (
|
||||
self.Meta.model.objects.exclude(
|
||||
pk=self.instance.pk,
|
||||
)
|
||||
if self.instance is not None
|
||||
else self.Meta.model.objects.all()
|
||||
)
|
||||
if ("name" in attrs) and objects.filter(
|
||||
name=name,
|
||||
).exists():
|
||||
raise serializers.ValidationError(
|
||||
{"error": "Object violates name unique constraint"},
|
||||
)
|
||||
if (
|
||||
"data_type" in attrs
|
||||
and attrs["data_type"] == CustomField.FieldDataType.SELECT
|
||||
) or (
|
||||
self.instance
|
||||
and self.instance.data_type == CustomField.FieldDataType.SELECT
|
||||
):
|
||||
if (
|
||||
"extra_data" not in attrs
|
||||
or "select_options" not in attrs["extra_data"]
|
||||
or not isinstance(attrs["extra_data"]["select_options"], list)
|
||||
or len(attrs["extra_data"]["select_options"]) == 0
|
||||
or not all(
|
||||
len(option.get("label", "")) > 0
|
||||
for option in attrs["extra_data"]["select_options"]
|
||||
)
|
||||
):
|
||||
raise serializers.ValidationError(
|
||||
{"error": "extra_data.select_options must be a valid list"},
|
||||
)
|
||||
# labels are valid, generate ids if not present
|
||||
for option in attrs["extra_data"]["select_options"]:
|
||||
if option.get("id") is None:
|
||||
option["id"] = get_random_string(length=16)
|
||||
elif (
|
||||
"data_type" in attrs
|
||||
and attrs["data_type"] == CustomField.FieldDataType.MONETARY
|
||||
and "extra_data" in attrs
|
||||
and "default_currency" in attrs["extra_data"]
|
||||
and attrs["extra_data"]["default_currency"] is not None
|
||||
and (
|
||||
not isinstance(attrs["extra_data"]["default_currency"], str)
|
||||
or (
|
||||
len(attrs["extra_data"]["default_currency"]) > 0
|
||||
and len(attrs["extra_data"]["default_currency"]) != 3
|
||||
)
|
||||
)
|
||||
):
|
||||
raise serializers.ValidationError(
|
||||
{"error": "extra_data.default_currency must be a 3-character string"},
|
||||
)
|
||||
return super().validate(attrs)
|
||||
|
||||
|
||||
def validate_documentlink_targets(user, doc_ids):
|
||||
if Document.objects.filter(id__in=doc_ids).count() != len(doc_ids):
|
||||
raise serializers.ValidationError(
|
||||
"Some documents in value don't exist or were specified twice.",
|
||||
)
|
||||
|
||||
if user is None:
|
||||
return
|
||||
|
||||
if (
|
||||
Document.objects.filter(id__in=doc_ids)
|
||||
.exclude(id__in=permitted_document_ids(user, perm="change_document"))
|
||||
.exists()
|
||||
):
|
||||
raise PermissionDenied(
|
||||
_("Insufficient permissions."),
|
||||
)
|
||||
|
||||
|
||||
class CustomFieldInstanceSerializer(serializers.ModelSerializer[CustomFieldInstance]):
|
||||
field = serializers.PrimaryKeyRelatedField(queryset=CustomField.objects.all())
|
||||
value = ReadWriteSerializerMethodField(allow_null=True)
|
||||
|
||||
def create(self, validated_data):
|
||||
# An instance is attached to a document
|
||||
document: Document = validated_data["document"]
|
||||
# And to a CustomField
|
||||
custom_field: CustomField = validated_data["field"]
|
||||
# This key must exist, as it is validated
|
||||
data_store_name = CustomFieldInstance.get_value_field_name(
|
||||
custom_field.data_type,
|
||||
)
|
||||
|
||||
if custom_field.data_type == CustomField.FieldDataType.DOCUMENTLINK:
|
||||
# prior to update so we can look for any docs that are going to be removed
|
||||
bulk_edit.reflect_doclinks(document, custom_field, validated_data["value"])
|
||||
|
||||
# Actually update or create the instance, providing the value
|
||||
# to fill in the correct attribute based on the type
|
||||
instance, _ = CustomFieldInstance.objects.update_or_create(
|
||||
document=document,
|
||||
field=custom_field,
|
||||
defaults={data_store_name: validated_data["value"]},
|
||||
)
|
||||
return instance
|
||||
|
||||
def get_value(self, obj: CustomFieldInstance) -> str | int | float | dict | None:
|
||||
return obj.value
|
||||
|
||||
def validate(self, data):
|
||||
"""
|
||||
Probably because we're kind of doing it odd, validation from the model
|
||||
doesn't run against the field "value", so we have to re-create it here.
|
||||
|
||||
Don't like it, but it is better than returning an HTTP 500 when the database
|
||||
hates the value
|
||||
"""
|
||||
data = super().validate(data)
|
||||
field: CustomField = data["field"]
|
||||
if "value" in data and data["value"] is not None:
|
||||
if (
|
||||
field.data_type == CustomField.FieldDataType.URL
|
||||
and len(data["value"]) > 0
|
||||
):
|
||||
uri_validator(data["value"])
|
||||
elif field.data_type == CustomField.FieldDataType.INT:
|
||||
integer_validator(data["value"])
|
||||
try:
|
||||
value_int = int(data["value"])
|
||||
except (TypeError, ValueError):
|
||||
raise serializers.ValidationError("Enter a valid integer.")
|
||||
# Keep values within the PostgreSQL integer range
|
||||
MinValueValidator(-2147483648)(value_int)
|
||||
MaxValueValidator(2147483647)(value_int)
|
||||
elif (
|
||||
field.data_type == CustomField.FieldDataType.MONETARY
|
||||
and data["value"] != ""
|
||||
):
|
||||
try:
|
||||
# First try to validate as a number from legacy format
|
||||
DecimalValidator(max_digits=12, decimal_places=2)(
|
||||
Decimal(str(data["value"])),
|
||||
)
|
||||
except Exception:
|
||||
# If that fails, try to validate as a monetary string
|
||||
RegexValidator(
|
||||
regex=r"^[A-Z]{3}-?\d+(\.\d{1,2})$",
|
||||
message="Must be a two-decimal number with optional currency code e.g. GBP123.45",
|
||||
)(data["value"])
|
||||
elif field.data_type == CustomField.FieldDataType.STRING:
|
||||
MaxLengthValidator(limit_value=128)(data["value"])
|
||||
elif field.data_type == CustomField.FieldDataType.SELECT:
|
||||
select_options = field.extra_data["select_options"]
|
||||
try:
|
||||
next(
|
||||
option
|
||||
for option in select_options
|
||||
if option["id"] == data["value"]
|
||||
)
|
||||
except Exception:
|
||||
raise serializers.ValidationError(
|
||||
f"Value must be an id of an element in {select_options}",
|
||||
)
|
||||
elif field.data_type == CustomField.FieldDataType.DOCUMENTLINK:
|
||||
if not (isinstance(data["value"], list) or data["value"] is None):
|
||||
raise serializers.ValidationError(
|
||||
"Value must be a list",
|
||||
)
|
||||
doc_ids = data["value"]
|
||||
request = self.context.get("request")
|
||||
validate_documentlink_targets(
|
||||
getattr(request, "user", None) if request is not None else None,
|
||||
doc_ids,
|
||||
)
|
||||
elif field.data_type == CustomField.FieldDataType.DATE:
|
||||
data["value"] = serializers.DateField().to_internal_value(data["value"])
|
||||
|
||||
return data
|
||||
|
||||
class Meta:
|
||||
model = CustomFieldInstance
|
||||
fields = [
|
||||
"value",
|
||||
"field",
|
||||
]
|
||||
|
||||
|
||||
class StoragePathSerializer(MatchingModelSerializer, OwnedObjectSerializer):
|
||||
class Meta:
|
||||
model = StoragePath
|
||||
fields = (
|
||||
"id",
|
||||
"slug",
|
||||
"name",
|
||||
"path",
|
||||
"match",
|
||||
"matching_algorithm",
|
||||
"is_insensitive",
|
||||
"document_count",
|
||||
"owner",
|
||||
"permissions",
|
||||
"user_can_change",
|
||||
"set_permissions",
|
||||
)
|
||||
|
||||
def validate_path(self, path: str):
|
||||
converted_path = convert_format_str_to_template_format(path)
|
||||
if converted_path != path:
|
||||
logger.warning(
|
||||
f"Storage path {path} is not using the new style format, consider updating",
|
||||
)
|
||||
result = validate_filepath_template_and_render(converted_path)
|
||||
|
||||
if result is None:
|
||||
raise serializers.ValidationError(_("Invalid variable detected."))
|
||||
|
||||
return converted_path
|
||||
|
||||
def update(self, instance, validated_data):
|
||||
"""
|
||||
When a storage path is updated, see if documents
|
||||
using it require a rename/move
|
||||
"""
|
||||
doc_ids = [doc.id for doc in instance.documents.all()]
|
||||
if doc_ids:
|
||||
bulk_edit.bulk_update_documents.apply_async(
|
||||
kwargs={"document_ids": doc_ids},
|
||||
headers={"trigger_source": PaperlessTask.TriggerSource.SYSTEM},
|
||||
)
|
||||
|
||||
return super().update(instance, validated_data)
|
||||
|
||||
|
||||
class StoragePathTestSerializer(SerializerWithPerms):
|
||||
path = serializers.CharField(
|
||||
required=True,
|
||||
label="Path",
|
||||
write_only=True,
|
||||
)
|
||||
|
||||
document = serializers.PrimaryKeyRelatedField(
|
||||
queryset=Document.objects.none(),
|
||||
required=True,
|
||||
label="Document",
|
||||
write_only=True,
|
||||
)
|
||||
|
||||
def __init__(self, *args: Any, **kwargs: Any) -> None:
|
||||
super().__init__(*args, **kwargs)
|
||||
request = self.context.get("request")
|
||||
user = getattr(request, "user", None) if request else None
|
||||
if user is not None and user.is_authenticated:
|
||||
document_field = self.fields.get("document")
|
||||
if not isinstance(document_field, serializers.PrimaryKeyRelatedField):
|
||||
return
|
||||
document_field.queryset = Document.objects.filter(
|
||||
id__in=permitted_document_ids(user),
|
||||
)
|
||||
@@ -0,0 +1,234 @@
|
||||
from __future__ import annotations
|
||||
|
||||
import logging
|
||||
import re
|
||||
|
||||
from django.conf import settings
|
||||
from rest_framework import serializers
|
||||
|
||||
from documents.models import CustomField
|
||||
from documents.models import SavedView
|
||||
from documents.models import SavedViewFilterRule
|
||||
from documents.models import UiSettings
|
||||
|
||||
from .base import OwnedObjectSerializer
|
||||
|
||||
logger = logging.getLogger("paperless.serializers")
|
||||
|
||||
|
||||
class SavedViewFilterRuleSerializer(serializers.ModelSerializer[SavedViewFilterRule]):
|
||||
class Meta:
|
||||
model = SavedViewFilterRule
|
||||
fields = ["rule_type", "value"]
|
||||
|
||||
|
||||
class SavedViewSerializer(OwnedObjectSerializer):
|
||||
filter_rules = SavedViewFilterRuleSerializer(many=True)
|
||||
|
||||
class Meta:
|
||||
model = SavedView
|
||||
fields = [
|
||||
"id",
|
||||
"name",
|
||||
"icon",
|
||||
"sort_field",
|
||||
"sort_reverse",
|
||||
"filter_rules",
|
||||
"page_size",
|
||||
"display_mode",
|
||||
"display_fields",
|
||||
"owner",
|
||||
"permissions",
|
||||
"user_can_change",
|
||||
"set_permissions",
|
||||
]
|
||||
|
||||
def _get_api_version(self) -> int:
|
||||
request = self.context.get("request")
|
||||
return int(
|
||||
request.version if request else settings.REST_FRAMEWORK["DEFAULT_VERSION"],
|
||||
)
|
||||
|
||||
def _update_legacy_visibility_preferences(
|
||||
self,
|
||||
saved_view_id: int,
|
||||
*,
|
||||
show_on_dashboard: bool | None,
|
||||
show_in_sidebar: bool | None,
|
||||
) -> UiSettings | None:
|
||||
if show_on_dashboard is None and show_in_sidebar is None:
|
||||
return None
|
||||
|
||||
request = self.context.get("request")
|
||||
user = request.user if request else self.user
|
||||
if user is None:
|
||||
return None
|
||||
|
||||
ui_settings, _ = UiSettings.objects.get_or_create(
|
||||
user=user,
|
||||
defaults={"settings": {}},
|
||||
)
|
||||
current_settings = (
|
||||
ui_settings.settings if isinstance(ui_settings.settings, dict) else {}
|
||||
)
|
||||
current_settings = dict(current_settings)
|
||||
|
||||
saved_views_settings = current_settings.get("saved_views")
|
||||
if isinstance(saved_views_settings, dict):
|
||||
saved_views_settings = dict(saved_views_settings)
|
||||
else:
|
||||
saved_views_settings = {}
|
||||
|
||||
dashboard_ids = {
|
||||
int(raw_id)
|
||||
for raw_id in saved_views_settings.get("dashboard_views_visible_ids", [])
|
||||
if str(raw_id).isdigit()
|
||||
}
|
||||
sidebar_ids = {
|
||||
int(raw_id)
|
||||
for raw_id in saved_views_settings.get("sidebar_views_visible_ids", [])
|
||||
if str(raw_id).isdigit()
|
||||
}
|
||||
|
||||
if show_on_dashboard is not None:
|
||||
if show_on_dashboard:
|
||||
dashboard_ids.add(saved_view_id)
|
||||
else:
|
||||
dashboard_ids.discard(saved_view_id)
|
||||
if show_in_sidebar is not None:
|
||||
if show_in_sidebar:
|
||||
sidebar_ids.add(saved_view_id)
|
||||
else:
|
||||
sidebar_ids.discard(saved_view_id)
|
||||
|
||||
saved_views_settings["dashboard_views_visible_ids"] = sorted(dashboard_ids)
|
||||
saved_views_settings["sidebar_views_visible_ids"] = sorted(sidebar_ids)
|
||||
current_settings["saved_views"] = saved_views_settings
|
||||
ui_settings.settings = current_settings
|
||||
ui_settings.save(update_fields=["settings"])
|
||||
return ui_settings
|
||||
|
||||
def to_representation(self, instance):
|
||||
# TODO: remove this and related backwards compatibility code when API v9 is dropped
|
||||
ret = super().to_representation(instance)
|
||||
request = self.context.get("request")
|
||||
api_version = self._get_api_version()
|
||||
|
||||
if api_version < 10:
|
||||
dashboard_ids = set()
|
||||
sidebar_ids = set()
|
||||
user = request.user if request else None
|
||||
if user is not None and hasattr(user, "ui_settings"):
|
||||
ui_settings = user.ui_settings.settings or None
|
||||
saved_views = None
|
||||
if isinstance(ui_settings, dict):
|
||||
saved_views = ui_settings.get("saved_views", {})
|
||||
if isinstance(saved_views, dict):
|
||||
dashboard_ids = set(
|
||||
saved_views.get("dashboard_views_visible_ids", []),
|
||||
)
|
||||
sidebar_ids = set(
|
||||
saved_views.get("sidebar_views_visible_ids", []),
|
||||
)
|
||||
ret["show_on_dashboard"] = instance.id in dashboard_ids
|
||||
ret["show_in_sidebar"] = instance.id in sidebar_ids
|
||||
|
||||
return ret
|
||||
|
||||
def to_internal_value(self, data):
|
||||
# TODO: remove this and related backwards compatibility code when API v9 is dropped
|
||||
api_version = self._get_api_version()
|
||||
if api_version >= 10:
|
||||
return super().to_internal_value(data)
|
||||
|
||||
normalized_data = data.copy()
|
||||
legacy_visibility_fields = {}
|
||||
boolean_field = serializers.BooleanField()
|
||||
|
||||
for field_name in ("show_on_dashboard", "show_in_sidebar"):
|
||||
if field_name in normalized_data:
|
||||
try:
|
||||
legacy_visibility_fields[field_name] = (
|
||||
boolean_field.to_internal_value(
|
||||
normalized_data.get(field_name),
|
||||
)
|
||||
)
|
||||
except serializers.ValidationError as exc:
|
||||
raise serializers.ValidationError({field_name: exc.detail})
|
||||
del normalized_data[field_name]
|
||||
|
||||
ret = super().to_internal_value(normalized_data)
|
||||
ret.update(legacy_visibility_fields)
|
||||
return ret
|
||||
|
||||
def validate(self, attrs):
|
||||
attrs = super().validate(attrs)
|
||||
if "display_fields" in attrs and attrs["display_fields"] is not None:
|
||||
for field in attrs["display_fields"]:
|
||||
if (
|
||||
SavedView.DisplayFields.CUSTOM_FIELD[:-2] in field
|
||||
): # i.e. check for 'custom_field_' prefix
|
||||
field_id = int(re.search(r"\d+", field)[0])
|
||||
if not CustomField.objects.filter(id=field_id).exists():
|
||||
raise serializers.ValidationError(
|
||||
f"Invalid field: {field}",
|
||||
)
|
||||
elif field not in SavedView.DisplayFields.values:
|
||||
raise serializers.ValidationError(
|
||||
f"Invalid field: {field}",
|
||||
)
|
||||
return attrs
|
||||
|
||||
def update(self, instance, validated_data):
|
||||
request = self.context.get("request")
|
||||
show_on_dashboard = validated_data.pop("show_on_dashboard", None)
|
||||
show_in_sidebar = validated_data.pop("show_in_sidebar", None)
|
||||
if "filter_rules" in validated_data:
|
||||
rules_data = validated_data.pop("filter_rules")
|
||||
else:
|
||||
rules_data = None
|
||||
if "user" in validated_data:
|
||||
# backwards compatibility
|
||||
validated_data["owner"] = validated_data.pop("user")
|
||||
if (
|
||||
"display_fields" in validated_data
|
||||
and isinstance(
|
||||
validated_data["display_fields"],
|
||||
list,
|
||||
)
|
||||
and len(validated_data["display_fields"]) == 0
|
||||
):
|
||||
validated_data["display_fields"] = None
|
||||
instance = super().update(instance, validated_data)
|
||||
if rules_data is not None:
|
||||
SavedViewFilterRule.objects.filter(saved_view=instance).delete()
|
||||
for rule_data in rules_data:
|
||||
SavedViewFilterRule.objects.create(saved_view=instance, **rule_data)
|
||||
ui_settings = self._update_legacy_visibility_preferences(
|
||||
instance.id,
|
||||
show_on_dashboard=show_on_dashboard,
|
||||
show_in_sidebar=show_in_sidebar,
|
||||
)
|
||||
if request is not None and ui_settings is not None:
|
||||
request.user.ui_settings = ui_settings
|
||||
return instance
|
||||
|
||||
def create(self, validated_data):
|
||||
request = self.context.get("request")
|
||||
show_on_dashboard = validated_data.pop("show_on_dashboard", None)
|
||||
show_in_sidebar = validated_data.pop("show_in_sidebar", None)
|
||||
rules_data = validated_data.pop("filter_rules")
|
||||
if "user" in validated_data:
|
||||
# backwards compatibility
|
||||
validated_data["owner"] = validated_data.pop("user")
|
||||
saved_view = super().create(validated_data)
|
||||
for rule_data in rules_data:
|
||||
SavedViewFilterRule.objects.create(saved_view=saved_view, **rule_data)
|
||||
ui_settings = self._update_legacy_visibility_preferences(
|
||||
saved_view.id,
|
||||
show_on_dashboard=show_on_dashboard,
|
||||
show_in_sidebar=show_in_sidebar,
|
||||
)
|
||||
if request is not None and ui_settings is not None:
|
||||
request.user.ui_settings = ui_settings
|
||||
return saved_view
|
||||
@@ -0,0 +1,204 @@
|
||||
from __future__ import annotations
|
||||
|
||||
import logging
|
||||
from datetime import timedelta
|
||||
|
||||
from django.core.exceptions import ValidationError
|
||||
from django.core.validators import EmailValidator
|
||||
from django.utils import timezone
|
||||
from django.utils.crypto import get_random_string
|
||||
from django.utils.translation import gettext as _
|
||||
from rest_framework import serializers
|
||||
from rest_framework.exceptions import PermissionDenied
|
||||
from rest_framework.fields import SerializerMethodField
|
||||
|
||||
from documents.models import Document
|
||||
from documents.models import ShareLink
|
||||
from documents.models import ShareLinkBundle
|
||||
from documents.permissions import has_perms_owner_aware
|
||||
|
||||
from .base import DocumentListSerializer
|
||||
from .base import OwnedObjectSerializer
|
||||
|
||||
logger = logging.getLogger("paperless.serializers")
|
||||
|
||||
|
||||
class EmailSerializer(DocumentListSerializer):
|
||||
addresses = serializers.CharField(
|
||||
required=True,
|
||||
label="Email addresses",
|
||||
help_text="Comma-separated email addresses",
|
||||
)
|
||||
|
||||
subject = serializers.CharField(
|
||||
required=True,
|
||||
label="Email subject",
|
||||
)
|
||||
|
||||
message = serializers.CharField(
|
||||
required=True,
|
||||
label="Email message",
|
||||
)
|
||||
|
||||
use_archive_version = serializers.BooleanField(
|
||||
default=True,
|
||||
label="Use archive version",
|
||||
help_text="Use archive version of documents if available",
|
||||
)
|
||||
|
||||
def validate_addresses(self, addresses):
|
||||
address_list = [addr.strip() for addr in addresses.split(",")]
|
||||
if not address_list:
|
||||
raise serializers.ValidationError("At least one email address is required")
|
||||
|
||||
email_validator = EmailValidator()
|
||||
try:
|
||||
for address in address_list:
|
||||
email_validator(address)
|
||||
except ValidationError:
|
||||
raise serializers.ValidationError(f"Invalid email address: {address}")
|
||||
|
||||
return ",".join(address_list)
|
||||
|
||||
def validate_documents(self, documents):
|
||||
super().validate_documents(documents)
|
||||
if not documents:
|
||||
raise serializers.ValidationError("At least one document is required")
|
||||
|
||||
return documents
|
||||
|
||||
|
||||
class ShareLinkSerializer(OwnedObjectSerializer):
|
||||
document_title = serializers.CharField(
|
||||
source="document.title",
|
||||
read_only=True,
|
||||
)
|
||||
|
||||
class Meta:
|
||||
model = ShareLink
|
||||
fields = (
|
||||
"id",
|
||||
"created",
|
||||
"expiration",
|
||||
"slug",
|
||||
"document",
|
||||
"document_title",
|
||||
"file_version",
|
||||
)
|
||||
|
||||
def create(self, validated_data):
|
||||
validated_data["slug"] = get_random_string(50)
|
||||
return super().create(validated_data)
|
||||
|
||||
def validate_document(self, document):
|
||||
if (
|
||||
self.user is not None
|
||||
and self.user.has_perm("documents.view_document")
|
||||
and has_perms_owner_aware(
|
||||
self.user,
|
||||
"view_document",
|
||||
document,
|
||||
)
|
||||
):
|
||||
return document
|
||||
raise PermissionDenied(
|
||||
_("Insufficient permissions."),
|
||||
)
|
||||
|
||||
|
||||
class ShareLinkBundleSerializer(OwnedObjectSerializer):
|
||||
document_ids = serializers.ListField(
|
||||
child=serializers.IntegerField(min_value=1),
|
||||
allow_empty=False,
|
||||
write_only=True,
|
||||
)
|
||||
expiration_days = serializers.IntegerField(
|
||||
required=False,
|
||||
allow_null=True,
|
||||
min_value=1,
|
||||
write_only=True,
|
||||
)
|
||||
documents = serializers.PrimaryKeyRelatedField(
|
||||
many=True,
|
||||
read_only=True,
|
||||
)
|
||||
document_count = SerializerMethodField()
|
||||
|
||||
class Meta:
|
||||
model = ShareLinkBundle
|
||||
fields = (
|
||||
"id",
|
||||
"created",
|
||||
"expiration",
|
||||
"expiration_days",
|
||||
"slug",
|
||||
"file_version",
|
||||
"status",
|
||||
"size_bytes",
|
||||
"last_error",
|
||||
"built_at",
|
||||
"documents",
|
||||
"document_ids",
|
||||
"document_count",
|
||||
)
|
||||
read_only_fields = (
|
||||
"id",
|
||||
"created",
|
||||
"expiration",
|
||||
"slug",
|
||||
"status",
|
||||
"size_bytes",
|
||||
"last_error",
|
||||
"built_at",
|
||||
"documents",
|
||||
"document_count",
|
||||
)
|
||||
|
||||
def validate_document_ids(self, value):
|
||||
unique_ids = set(value)
|
||||
if len(unique_ids) != len(value):
|
||||
raise serializers.ValidationError(
|
||||
_("Duplicate document identifiers are not allowed."),
|
||||
)
|
||||
return value
|
||||
|
||||
def create(self, validated_data):
|
||||
document_ids = validated_data.pop("document_ids")
|
||||
expiration_days = validated_data.pop("expiration_days", None)
|
||||
validated_data["slug"] = get_random_string(50)
|
||||
if expiration_days:
|
||||
validated_data["expiration"] = timezone.now() + timedelta(
|
||||
days=expiration_days,
|
||||
)
|
||||
else:
|
||||
validated_data["expiration"] = None
|
||||
|
||||
share_link_bundle = super().create(validated_data)
|
||||
|
||||
documents = list(
|
||||
Document.objects.filter(pk__in=document_ids).only(
|
||||
"pk",
|
||||
),
|
||||
)
|
||||
documents_by_id = {doc.pk: doc for doc in documents}
|
||||
missing = [
|
||||
str(doc_id) for doc_id in document_ids if doc_id not in documents_by_id
|
||||
]
|
||||
if missing:
|
||||
raise serializers.ValidationError(
|
||||
{
|
||||
"document_ids": _(
|
||||
"Documents not found: %(ids)s",
|
||||
)
|
||||
% {"ids": ", ".join(missing)},
|
||||
},
|
||||
)
|
||||
|
||||
ordered_documents = [documents_by_id[doc_id] for doc_id in document_ids]
|
||||
share_link_bundle.documents.set(ordered_documents)
|
||||
share_link_bundle.document_total = len(ordered_documents)
|
||||
|
||||
return share_link_bundle
|
||||
|
||||
def get_document_count(self, obj: ShareLinkBundle) -> int:
|
||||
return getattr(obj, "document_total") or obj.documents.count()
|
||||
@@ -0,0 +1,63 @@
|
||||
from __future__ import annotations
|
||||
|
||||
import logging
|
||||
|
||||
from rest_framework import serializers
|
||||
|
||||
from documents.models import Document
|
||||
from documents.models import UiSettings
|
||||
|
||||
from .base import SerializerWithPerms
|
||||
|
||||
logger = logging.getLogger("paperless.serializers")
|
||||
|
||||
|
||||
class UiSettingsViewSerializer(serializers.ModelSerializer[UiSettings]):
|
||||
settings = serializers.DictField(required=False, allow_null=True)
|
||||
|
||||
class Meta:
|
||||
model = UiSettings
|
||||
depth = 1
|
||||
fields = [
|
||||
"id",
|
||||
"settings",
|
||||
]
|
||||
|
||||
def validate_settings(self, settings):
|
||||
# we never save update checking backend setting
|
||||
if "update_checking" in settings:
|
||||
try:
|
||||
settings["update_checking"].pop("backend_setting")
|
||||
except KeyError:
|
||||
pass
|
||||
return settings
|
||||
|
||||
def create(self, validated_data):
|
||||
ui_settings = UiSettings.objects.update_or_create(
|
||||
user=validated_data.get("user"),
|
||||
defaults={"settings": validated_data.get("settings", None)},
|
||||
)
|
||||
return ui_settings
|
||||
|
||||
|
||||
class TrashSerializer(SerializerWithPerms):
|
||||
documents = serializers.ListField(
|
||||
required=False,
|
||||
label="Documents",
|
||||
write_only=True,
|
||||
child=serializers.IntegerField(),
|
||||
)
|
||||
|
||||
action = serializers.ChoiceField(
|
||||
choices=["restore", "empty"],
|
||||
label="Action",
|
||||
write_only=True,
|
||||
)
|
||||
|
||||
def validate_documents(self, documents: list[int]) -> list[int]:
|
||||
count = Document.deleted_objects.filter(id__in=documents).count()
|
||||
if not count == len(documents):
|
||||
raise serializers.ValidationError(
|
||||
"Some documents in the list have not yet been deleted.",
|
||||
)
|
||||
return documents
|
||||
@@ -0,0 +1,246 @@
|
||||
from __future__ import annotations
|
||||
|
||||
import logging
|
||||
from typing import Any
|
||||
|
||||
from rest_framework import serializers
|
||||
|
||||
from documents.models import Document
|
||||
from documents.models import PaperlessTask
|
||||
from documents.permissions import permitted_document_ids
|
||||
|
||||
from .base import OwnedObjectSerializer
|
||||
|
||||
logger = logging.getLogger("paperless.serializers")
|
||||
|
||||
|
||||
class TaskSerializerV10(OwnedObjectSerializer):
|
||||
"""Task serializer for API v10+ using new field names."""
|
||||
|
||||
related_document_ids = serializers.ListField(
|
||||
child=serializers.IntegerField(),
|
||||
read_only=True,
|
||||
)
|
||||
task_type_display = serializers.CharField(
|
||||
source="get_task_type_display",
|
||||
read_only=True,
|
||||
)
|
||||
trigger_source_display = serializers.CharField(
|
||||
source="get_trigger_source_display",
|
||||
read_only=True,
|
||||
)
|
||||
status_display = serializers.CharField(
|
||||
source="get_status_display",
|
||||
read_only=True,
|
||||
)
|
||||
|
||||
class Meta:
|
||||
model = PaperlessTask
|
||||
fields = (
|
||||
"id",
|
||||
"task_id",
|
||||
"task_type",
|
||||
"task_type_display",
|
||||
"trigger_source",
|
||||
"trigger_source_display",
|
||||
"status",
|
||||
"status_display",
|
||||
"date_created",
|
||||
"date_started",
|
||||
"date_done",
|
||||
"duration_seconds",
|
||||
"wait_time_seconds",
|
||||
"input_data",
|
||||
"result_data",
|
||||
"related_document_ids",
|
||||
"acknowledged",
|
||||
"owner",
|
||||
)
|
||||
read_only_fields = fields
|
||||
|
||||
|
||||
class TaskSerializerV9(serializers.ModelSerializer[PaperlessTask]):
|
||||
"""Task serializer for API v9 backwards compatibility.
|
||||
|
||||
Maps old field names to the new model fields so existing clients continue
|
||||
to work unchanged.
|
||||
"""
|
||||
|
||||
# v9 field: task_name -> task_type (with value remapping for renamed tasks)
|
||||
task_name = serializers.SerializerMethodField()
|
||||
|
||||
# v9 field: task_file_name -> input_data.filename
|
||||
task_file_name = serializers.SerializerMethodField()
|
||||
|
||||
# v9 field: type -> trigger_source (mapped to old enum labels)
|
||||
type = serializers.SerializerMethodField()
|
||||
|
||||
# v9 field: status -> uppercase Celery state strings
|
||||
status = serializers.SerializerMethodField()
|
||||
|
||||
# v9 field: result -> derived from result_data
|
||||
result = serializers.SerializerMethodField()
|
||||
|
||||
# v9 field: related_document -> first document ID from result_data
|
||||
related_document = serializers.SerializerMethodField()
|
||||
|
||||
# v9 field: duplicate_documents -> list of duplicate IDs from result_data
|
||||
duplicate_documents = serializers.SerializerMethodField()
|
||||
|
||||
class Meta:
|
||||
model = PaperlessTask
|
||||
fields = (
|
||||
"id",
|
||||
"task_id",
|
||||
"task_name",
|
||||
"task_file_name",
|
||||
"type",
|
||||
"status",
|
||||
"date_created",
|
||||
"date_done",
|
||||
"result",
|
||||
"acknowledged",
|
||||
"related_document",
|
||||
"duplicate_documents",
|
||||
"owner",
|
||||
)
|
||||
read_only_fields = fields
|
||||
|
||||
_TASK_TYPE_TO_V9_NAME = {
|
||||
PaperlessTask.TaskType.SANITY_CHECK: "check_sanity",
|
||||
PaperlessTask.TaskType.LLM_INDEX: "llmindex_update",
|
||||
}
|
||||
|
||||
def get_result(self, obj: PaperlessTask) -> str | None:
|
||||
"""Reconstruct a human-readable result string from result_data for v9 clients."""
|
||||
if not obj.result_data:
|
||||
return None
|
||||
if doc_id := obj.result_data.get("document_id"):
|
||||
return f"Success. New document id {doc_id} created"
|
||||
if reason := obj.result_data.get("reason"):
|
||||
return reason
|
||||
if dup_id := obj.result_data.get("duplicate_of"):
|
||||
return f"Not consuming: It is a duplicate of document #{dup_id}"
|
||||
if error := obj.result_data.get("error_message"):
|
||||
return error
|
||||
return None
|
||||
|
||||
def get_task_name(self, obj: PaperlessTask) -> str:
|
||||
return self._TASK_TYPE_TO_V9_NAME.get(obj.task_type, obj.task_type)
|
||||
|
||||
def get_task_file_name(self, obj: PaperlessTask) -> str | None:
|
||||
if not obj.input_data:
|
||||
return None
|
||||
return obj.input_data.get("filename")
|
||||
|
||||
_STATUS_TO_V9 = {
|
||||
PaperlessTask.Status.PENDING: "PENDING",
|
||||
PaperlessTask.Status.STARTED: "STARTED",
|
||||
PaperlessTask.Status.SUCCESS: "SUCCESS",
|
||||
PaperlessTask.Status.FAILURE: "FAILURE",
|
||||
PaperlessTask.Status.REVOKED: "REVOKED",
|
||||
}
|
||||
|
||||
def get_status(self, obj: PaperlessTask) -> str:
|
||||
return self._STATUS_TO_V9.get(obj.status, obj.status.upper())
|
||||
|
||||
_TRIGGER_SOURCE_TO_V9_TYPE = {
|
||||
PaperlessTask.TriggerSource.SCHEDULED: "scheduled_task",
|
||||
PaperlessTask.TriggerSource.SYSTEM: "auto_task",
|
||||
# Email and folder-consumer documents are system-initiated, not manually triggered
|
||||
PaperlessTask.TriggerSource.EMAIL_CONSUME: "auto_task",
|
||||
PaperlessTask.TriggerSource.FOLDER_CONSUME: "auto_task",
|
||||
}
|
||||
|
||||
def get_type(self, obj: PaperlessTask) -> str:
|
||||
return self._TRIGGER_SOURCE_TO_V9_TYPE.get(obj.trigger_source, "manual_task")
|
||||
|
||||
def get_related_document(self, obj: PaperlessTask) -> int | None:
|
||||
ids = obj.related_document_ids
|
||||
return ids[0] if ids else None
|
||||
|
||||
def get_duplicate_documents(
|
||||
self,
|
||||
obj: PaperlessTask,
|
||||
) -> list[dict[str, Any]]:
|
||||
if not obj.result_data:
|
||||
return []
|
||||
dup_of = obj.result_data.get("duplicate_of")
|
||||
if dup_of is None:
|
||||
return []
|
||||
request = self.context.get("request")
|
||||
if request is None:
|
||||
return []
|
||||
user = request.user
|
||||
qs = Document.global_objects.filter(pk=dup_of)
|
||||
if not user.is_staff:
|
||||
allowed_ids = permitted_document_ids(user, include_deleted=True)
|
||||
qs = qs.filter(pk__in=allowed_ids)
|
||||
return list(qs.values("id", "title", "deleted_at"))
|
||||
|
||||
|
||||
class TaskSummarySerializer(serializers.Serializer[dict[str, Any]]):
|
||||
task_type = serializers.CharField()
|
||||
total_count = serializers.IntegerField()
|
||||
pending_count = serializers.IntegerField()
|
||||
success_count = serializers.IntegerField()
|
||||
failure_count = serializers.IntegerField()
|
||||
avg_duration_seconds = serializers.FloatField(allow_null=True)
|
||||
avg_wait_time_seconds = serializers.FloatField(allow_null=True)
|
||||
last_run = serializers.DateTimeField(allow_null=True)
|
||||
last_success = serializers.DateTimeField(allow_null=True)
|
||||
last_failure = serializers.DateTimeField(allow_null=True)
|
||||
|
||||
|
||||
class RunTaskSerializer(serializers.Serializer[dict[str, str]]):
|
||||
task_type = serializers.ChoiceField(
|
||||
choices=PaperlessTask.TaskType.choices,
|
||||
label="Task Type",
|
||||
write_only=True,
|
||||
)
|
||||
|
||||
|
||||
class AcknowledgeTasksViewSerializer(serializers.Serializer[dict[str, Any]]):
|
||||
tasks = serializers.ListField(
|
||||
required=False,
|
||||
label="Tasks",
|
||||
write_only=True,
|
||||
child=serializers.IntegerField(),
|
||||
)
|
||||
all = serializers.BooleanField(
|
||||
required=False,
|
||||
default=False,
|
||||
label="All",
|
||||
write_only=True,
|
||||
)
|
||||
|
||||
def _validate_task_id_list(self, tasks, name="tasks") -> None:
|
||||
if not isinstance(tasks, list):
|
||||
raise serializers.ValidationError(f"{name} must be a list")
|
||||
if not all(isinstance(i, int) for i in tasks):
|
||||
raise serializers.ValidationError(f"{name} must be a list of integers")
|
||||
queryset = self.context.get("queryset", PaperlessTask.objects.all())
|
||||
count = queryset.filter(id__in=tasks).count()
|
||||
if not count == len(tasks):
|
||||
raise serializers.ValidationError(
|
||||
f"Some tasks in {name} don't exist or were specified twice.",
|
||||
)
|
||||
|
||||
def validate_tasks(self, tasks):
|
||||
self._validate_task_id_list(tasks)
|
||||
return tasks
|
||||
|
||||
def validate(self, attrs):
|
||||
acknowledge_all = attrs.get("all", False)
|
||||
task_ids = attrs.get("tasks")
|
||||
|
||||
if acknowledge_all and task_ids is not None:
|
||||
raise serializers.ValidationError(
|
||||
"Set either all or tasks, not both.",
|
||||
)
|
||||
if not acknowledge_all and task_ids is None:
|
||||
raise serializers.ValidationError(
|
||||
"Either all must be true or tasks must be provided.",
|
||||
)
|
||||
|
||||
return attrs
|
||||
@@ -0,0 +1,199 @@
|
||||
from __future__ import annotations
|
||||
|
||||
import logging
|
||||
from datetime import datetime
|
||||
from typing import Any
|
||||
|
||||
import magic
|
||||
from django.conf import settings
|
||||
from django.utils.translation import gettext as _
|
||||
from rest_framework import serializers
|
||||
|
||||
from documents.models import Correspondent
|
||||
from documents.models import CustomField
|
||||
from documents.models import Document
|
||||
from documents.models import DocumentType
|
||||
from documents.models import StoragePath
|
||||
from documents.models import Tag
|
||||
from documents.parsers import is_mime_type_supported
|
||||
|
||||
from .metadata import CustomFieldInstanceSerializer
|
||||
|
||||
logger = logging.getLogger("paperless.serializers")
|
||||
|
||||
|
||||
class PostDocumentSerializer(serializers.Serializer[dict[str, Any]]):
|
||||
created = serializers.DateTimeField(
|
||||
label="Created",
|
||||
allow_null=True,
|
||||
write_only=True,
|
||||
required=False,
|
||||
)
|
||||
|
||||
document = serializers.FileField(
|
||||
label="Document",
|
||||
write_only=True,
|
||||
)
|
||||
|
||||
title = serializers.CharField(
|
||||
label="Title",
|
||||
write_only=True,
|
||||
required=False,
|
||||
)
|
||||
|
||||
correspondent = serializers.PrimaryKeyRelatedField(
|
||||
queryset=Correspondent.objects.all(),
|
||||
label="Correspondent",
|
||||
allow_null=True,
|
||||
write_only=True,
|
||||
required=False,
|
||||
)
|
||||
|
||||
document_type = serializers.PrimaryKeyRelatedField(
|
||||
queryset=DocumentType.objects.all(),
|
||||
label="Document type",
|
||||
allow_null=True,
|
||||
write_only=True,
|
||||
required=False,
|
||||
)
|
||||
|
||||
storage_path = serializers.PrimaryKeyRelatedField(
|
||||
queryset=StoragePath.objects.all(),
|
||||
label="Storage path",
|
||||
allow_null=True,
|
||||
write_only=True,
|
||||
required=False,
|
||||
)
|
||||
|
||||
tags = serializers.PrimaryKeyRelatedField(
|
||||
many=True,
|
||||
queryset=Tag.objects.all(),
|
||||
label="Tags",
|
||||
write_only=True,
|
||||
required=False,
|
||||
)
|
||||
|
||||
archive_serial_number = serializers.IntegerField(
|
||||
label="ASN",
|
||||
write_only=True,
|
||||
required=False,
|
||||
min_value=Document.ARCHIVE_SERIAL_NUMBER_MIN,
|
||||
max_value=Document.ARCHIVE_SERIAL_NUMBER_MAX,
|
||||
)
|
||||
|
||||
# Accept either a list of custom field ids or a dict mapping id -> value
|
||||
custom_fields = serializers.JSONField(
|
||||
label="Custom fields",
|
||||
write_only=True,
|
||||
required=False,
|
||||
)
|
||||
|
||||
from_webui = serializers.BooleanField(
|
||||
label="Documents are from Paperless-ngx WebUI",
|
||||
write_only=True,
|
||||
required=False,
|
||||
)
|
||||
|
||||
def validate_document(self, document):
|
||||
document_data = document.file.read()
|
||||
mime_type = magic.from_buffer(document_data, mime=True)
|
||||
|
||||
if not is_mime_type_supported(mime_type):
|
||||
if (
|
||||
mime_type in settings.CONSUMER_PDF_RECOVERABLE_MIME_TYPES
|
||||
and document.name.endswith(
|
||||
".pdf",
|
||||
)
|
||||
):
|
||||
# If the file is an invalid PDF, we can try to recover it later in the consumer
|
||||
mime_type = "application/pdf"
|
||||
else:
|
||||
raise serializers.ValidationError(
|
||||
_("File type %(type)s not supported") % {"type": mime_type},
|
||||
)
|
||||
|
||||
return document.name, document_data
|
||||
|
||||
def validate_correspondent(self, correspondent):
|
||||
if correspondent:
|
||||
return correspondent.id
|
||||
else:
|
||||
return None
|
||||
|
||||
def validate_document_type(self, document_type):
|
||||
if document_type:
|
||||
return document_type.id
|
||||
else:
|
||||
return None
|
||||
|
||||
def validate_storage_path(self, storage_path):
|
||||
if storage_path:
|
||||
return storage_path.id
|
||||
else:
|
||||
return None
|
||||
|
||||
def validate_tags(self, tags):
|
||||
if tags:
|
||||
return [tag.id for tag in tags]
|
||||
else:
|
||||
return None
|
||||
|
||||
def validate_custom_fields(self, custom_fields):
|
||||
if not custom_fields:
|
||||
return None
|
||||
|
||||
# Normalize single values to a list
|
||||
if isinstance(custom_fields, int):
|
||||
custom_fields = [custom_fields]
|
||||
if isinstance(custom_fields, dict):
|
||||
custom_field_serializer = CustomFieldInstanceSerializer()
|
||||
normalized = {}
|
||||
for field_id, value in custom_fields.items():
|
||||
try:
|
||||
field_id_int = int(field_id)
|
||||
except (TypeError, ValueError):
|
||||
raise serializers.ValidationError(
|
||||
_("Custom field id must be an integer: %(id)s")
|
||||
% {"id": field_id},
|
||||
)
|
||||
try:
|
||||
field = CustomField.objects.get(id=field_id_int)
|
||||
except CustomField.DoesNotExist:
|
||||
raise serializers.ValidationError(
|
||||
_("Custom field with id %(id)s does not exist")
|
||||
% {"id": field_id_int},
|
||||
)
|
||||
custom_field_serializer.validate(
|
||||
{
|
||||
"field": field,
|
||||
"value": value,
|
||||
},
|
||||
)
|
||||
normalized[field_id_int] = value
|
||||
return normalized
|
||||
elif isinstance(custom_fields, list):
|
||||
try:
|
||||
ids = [int(i) for i in custom_fields]
|
||||
except (TypeError, ValueError):
|
||||
raise serializers.ValidationError(
|
||||
_(
|
||||
"Custom fields must be a list of integers or an object mapping ids to values.",
|
||||
),
|
||||
)
|
||||
if CustomField.objects.filter(id__in=ids).count() != len(set(ids)):
|
||||
raise serializers.ValidationError(
|
||||
_("Some custom fields don't exist or were specified twice."),
|
||||
)
|
||||
return ids
|
||||
raise serializers.ValidationError(
|
||||
_(
|
||||
"Custom fields must be a list of integers or an object mapping ids to values.",
|
||||
),
|
||||
)
|
||||
|
||||
# custom_fields_w_values handled via validate_custom_fields
|
||||
|
||||
def validate_created(self, created):
|
||||
# support datetime format for created for backwards compatibility
|
||||
if isinstance(created, datetime):
|
||||
return created.date()
|
||||
@@ -0,0 +1,584 @@
|
||||
from __future__ import annotations
|
||||
|
||||
import logging
|
||||
|
||||
from django.db.models import Count
|
||||
from rest_framework import fields
|
||||
from rest_framework import serializers
|
||||
|
||||
from documents.data_models import DocumentSource
|
||||
from documents.filters import CustomFieldQueryParser
|
||||
from documents.models import Workflow
|
||||
from documents.models import WorkflowAction
|
||||
from documents.models import WorkflowActionEmail
|
||||
from documents.models import WorkflowActionWebhook
|
||||
from documents.models import WorkflowTrigger
|
||||
from documents.templating.workflows import validate_workflow_template
|
||||
from documents.validators import url_validator
|
||||
|
||||
from .metadata import CorrespondentField
|
||||
from .metadata import DocumentTypeField
|
||||
from .metadata import StoragePathField
|
||||
from .metadata import TagsField
|
||||
|
||||
logger = logging.getLogger("paperless.serializers")
|
||||
|
||||
|
||||
class WorkflowTriggerSerializer(serializers.ModelSerializer[WorkflowTrigger]):
|
||||
id = serializers.IntegerField(required=False, allow_null=True)
|
||||
sources = fields.MultipleChoiceField(
|
||||
choices=WorkflowTrigger.DocumentSourceChoices.choices,
|
||||
allow_empty=True,
|
||||
default={
|
||||
DocumentSource.ConsumeFolder,
|
||||
DocumentSource.ApiUpload,
|
||||
DocumentSource.MailFetch,
|
||||
},
|
||||
)
|
||||
|
||||
type = serializers.ChoiceField(
|
||||
choices=WorkflowTrigger.WorkflowTriggerType.choices,
|
||||
label="Trigger Type",
|
||||
)
|
||||
|
||||
class Meta:
|
||||
model = WorkflowTrigger
|
||||
fields = [
|
||||
"id",
|
||||
"sources",
|
||||
"type",
|
||||
"filter_path",
|
||||
"filter_filename",
|
||||
"filter_mailrule",
|
||||
"matching_algorithm",
|
||||
"match",
|
||||
"is_insensitive",
|
||||
"filter_has_tags",
|
||||
"filter_has_all_tags",
|
||||
"filter_has_not_tags",
|
||||
"filter_custom_field_query",
|
||||
"filter_has_any_correspondents",
|
||||
"filter_has_not_correspondents",
|
||||
"filter_has_any_document_types",
|
||||
"filter_has_not_document_types",
|
||||
"filter_has_any_storage_paths",
|
||||
"filter_has_not_storage_paths",
|
||||
"filter_has_correspondent",
|
||||
"filter_has_document_type",
|
||||
"filter_has_storage_path",
|
||||
"schedule_offset_days",
|
||||
"schedule_is_recurring",
|
||||
"schedule_recurring_interval_days",
|
||||
"schedule_date_field",
|
||||
"schedule_date_custom_field",
|
||||
]
|
||||
|
||||
def validate(self, attrs):
|
||||
# Empty strings treated as None to avoid unexpected behavior
|
||||
if (
|
||||
"filter_filename" in attrs
|
||||
and attrs["filter_filename"] is not None
|
||||
and len(attrs["filter_filename"]) == 0
|
||||
):
|
||||
attrs["filter_filename"] = None
|
||||
if (
|
||||
"filter_path" in attrs
|
||||
and attrs["filter_path"] is not None
|
||||
and len(attrs["filter_path"]) == 0
|
||||
):
|
||||
attrs["filter_path"] = None
|
||||
|
||||
if (
|
||||
"filter_custom_field_query" in attrs
|
||||
and attrs["filter_custom_field_query"] is not None
|
||||
and len(attrs["filter_custom_field_query"]) == 0
|
||||
):
|
||||
attrs["filter_custom_field_query"] = None
|
||||
|
||||
if (
|
||||
"filter_custom_field_query" in attrs
|
||||
and attrs["filter_custom_field_query"] is not None
|
||||
):
|
||||
parser = CustomFieldQueryParser("filter_custom_field_query")
|
||||
parser.parse(attrs["filter_custom_field_query"])
|
||||
|
||||
trigger_type = attrs.get("type", getattr(self.instance, "type", None))
|
||||
if (
|
||||
trigger_type == WorkflowTrigger.WorkflowTriggerType.CONSUMPTION
|
||||
and "filter_mailrule" not in attrs
|
||||
and ("filter_filename" not in attrs or attrs["filter_filename"] is None)
|
||||
and ("filter_path" not in attrs or attrs["filter_path"] is None)
|
||||
):
|
||||
raise serializers.ValidationError(
|
||||
"File name, path or mail rule filter are required",
|
||||
)
|
||||
|
||||
return attrs
|
||||
|
||||
@staticmethod
|
||||
def normalize_workflow_trigger_sources(trigger) -> None:
|
||||
"""
|
||||
Convert sources to strings to handle django-multiselectfield v1.0 changes
|
||||
"""
|
||||
if trigger and "sources" in trigger:
|
||||
trigger["sources"] = [
|
||||
str(s.value if hasattr(s, "value") else s) for s in trigger["sources"]
|
||||
]
|
||||
|
||||
def create(self, validated_data):
|
||||
WorkflowTriggerSerializer.normalize_workflow_trigger_sources(validated_data)
|
||||
return super().create(validated_data)
|
||||
|
||||
def update(self, instance, validated_data):
|
||||
WorkflowTriggerSerializer.normalize_workflow_trigger_sources(validated_data)
|
||||
return super().update(instance, validated_data)
|
||||
|
||||
|
||||
class WorkflowActionEmailSerializer(serializers.ModelSerializer[WorkflowActionEmail]):
|
||||
id = serializers.IntegerField(allow_null=True, required=False)
|
||||
|
||||
class Meta:
|
||||
model = WorkflowActionEmail
|
||||
fields = [
|
||||
"id",
|
||||
"subject",
|
||||
"body",
|
||||
"to",
|
||||
"include_document",
|
||||
]
|
||||
|
||||
|
||||
class WorkflowActionWebhookSerializer(
|
||||
serializers.ModelSerializer[WorkflowActionWebhook],
|
||||
):
|
||||
id = serializers.IntegerField(allow_null=True, required=False)
|
||||
|
||||
def validate_url(self, url):
|
||||
url_validator(url)
|
||||
return url
|
||||
|
||||
class Meta:
|
||||
model = WorkflowActionWebhook
|
||||
fields = [
|
||||
"id",
|
||||
"url",
|
||||
"use_params",
|
||||
"as_json",
|
||||
"params",
|
||||
"body",
|
||||
"headers",
|
||||
"include_document",
|
||||
]
|
||||
|
||||
|
||||
class WorkflowActionSerializer(serializers.ModelSerializer[WorkflowAction]):
|
||||
id = serializers.IntegerField(required=False, allow_null=True)
|
||||
assign_correspondent = CorrespondentField(allow_null=True, required=False)
|
||||
assign_tags = TagsField(many=True, allow_null=True, required=False)
|
||||
assign_document_type = DocumentTypeField(allow_null=True, required=False)
|
||||
assign_storage_path = StoragePathField(allow_null=True, required=False)
|
||||
email = WorkflowActionEmailSerializer(allow_null=True, required=False)
|
||||
webhook = WorkflowActionWebhookSerializer(allow_null=True, required=False)
|
||||
|
||||
class Meta:
|
||||
model = WorkflowAction
|
||||
fields = [
|
||||
"id",
|
||||
"type",
|
||||
"assign_title",
|
||||
"assign_tags",
|
||||
"assign_correspondent",
|
||||
"assign_document_type",
|
||||
"assign_storage_path",
|
||||
"assign_owner",
|
||||
"assign_view_users",
|
||||
"assign_view_groups",
|
||||
"assign_change_users",
|
||||
"assign_change_groups",
|
||||
"assign_custom_fields",
|
||||
"assign_custom_fields_values",
|
||||
"remove_all_tags",
|
||||
"remove_tags",
|
||||
"remove_all_correspondents",
|
||||
"remove_correspondents",
|
||||
"remove_all_document_types",
|
||||
"remove_document_types",
|
||||
"remove_all_storage_paths",
|
||||
"remove_storage_paths",
|
||||
"remove_custom_fields",
|
||||
"remove_all_custom_fields",
|
||||
"remove_all_owners",
|
||||
"remove_owners",
|
||||
"remove_all_permissions",
|
||||
"remove_view_users",
|
||||
"remove_view_groups",
|
||||
"remove_change_users",
|
||||
"remove_change_groups",
|
||||
"email",
|
||||
"webhook",
|
||||
"passwords",
|
||||
"ai_suggestion_fields",
|
||||
"ai_create_missing",
|
||||
"ai_overwrite_existing",
|
||||
]
|
||||
|
||||
def validate(self, attrs):
|
||||
if "assign_title" in attrs and attrs["assign_title"] is not None:
|
||||
if len(attrs["assign_title"]) == 0:
|
||||
# Empty strings treated as None to avoid unexpected behavior
|
||||
attrs["assign_title"] = None
|
||||
else:
|
||||
try:
|
||||
validate_workflow_template(attrs["assign_title"])
|
||||
except (ValueError, KeyError) as e:
|
||||
raise serializers.ValidationError(
|
||||
{"assign_title": f"{e.args[0]}"},
|
||||
)
|
||||
|
||||
if attrs.get("assign_custom_fields_values"):
|
||||
# Empty strings treated as None to avoid unexpected behavior
|
||||
attrs["assign_custom_fields_values"] = {
|
||||
field_id: (None if value == "" else value)
|
||||
for field_id, value in attrs["assign_custom_fields_values"].items()
|
||||
}
|
||||
|
||||
if (
|
||||
"type" in attrs
|
||||
and attrs["type"] == WorkflowAction.WorkflowActionType.EMAIL
|
||||
and "email" not in attrs
|
||||
):
|
||||
raise serializers.ValidationError(
|
||||
"Email data is required for email actions",
|
||||
)
|
||||
|
||||
if (
|
||||
"type" in attrs
|
||||
and attrs["type"] == WorkflowAction.WorkflowActionType.WEBHOOK
|
||||
and "webhook" not in attrs
|
||||
):
|
||||
raise serializers.ValidationError(
|
||||
"Webhook data is required for webhook actions",
|
||||
)
|
||||
|
||||
if (
|
||||
"type" in attrs
|
||||
and attrs["type"] == WorkflowAction.WorkflowActionType.PASSWORD_REMOVAL
|
||||
):
|
||||
passwords = attrs.get("passwords")
|
||||
# ensure passwords is a non-empty list of non-empty strings
|
||||
if (
|
||||
passwords is None
|
||||
or not isinstance(passwords, list)
|
||||
or len(passwords) == 0
|
||||
or any(not isinstance(pw, str) for pw in passwords)
|
||||
or any(len(pw.strip()) == 0 for pw in passwords)
|
||||
):
|
||||
raise serializers.ValidationError(
|
||||
"Passwords are required for password removal actions",
|
||||
)
|
||||
|
||||
if (
|
||||
"type" in attrs
|
||||
and attrs["type"] == WorkflowAction.WorkflowActionType.APPLY_AI_SUGGESTIONS
|
||||
):
|
||||
fields = attrs.get("ai_suggestion_fields")
|
||||
valid_fields = set(WorkflowAction.AISuggestionField.values)
|
||||
if (
|
||||
fields is None
|
||||
or not isinstance(fields, list)
|
||||
or len(fields) == 0
|
||||
or any(field not in valid_fields for field in fields)
|
||||
):
|
||||
raise serializers.ValidationError(
|
||||
"At least one valid field is required for apply AI "
|
||||
f"suggestions actions, options are: {sorted(valid_fields)}",
|
||||
)
|
||||
|
||||
return attrs
|
||||
|
||||
|
||||
class WorkflowSerializer(serializers.ModelSerializer[Workflow]):
|
||||
order = serializers.IntegerField(required=False)
|
||||
|
||||
triggers = WorkflowTriggerSerializer(many=True)
|
||||
actions = WorkflowActionSerializer(many=True)
|
||||
|
||||
class Meta:
|
||||
model = Workflow
|
||||
fields = [
|
||||
"id",
|
||||
"name",
|
||||
"order",
|
||||
"enabled",
|
||||
"triggers",
|
||||
"actions",
|
||||
]
|
||||
|
||||
def validate(self, attrs):
|
||||
attrs = super().validate(attrs)
|
||||
|
||||
if "actions" in attrs:
|
||||
has_remote_ocr_action = any(
|
||||
action.get("type") == WorkflowAction.WorkflowActionType.REMOTE_OCR
|
||||
for action in attrs["actions"]
|
||||
)
|
||||
has_ai_suggestions_action = any(
|
||||
action.get("type")
|
||||
== WorkflowAction.WorkflowActionType.APPLY_AI_SUGGESTIONS
|
||||
for action in attrs["actions"]
|
||||
)
|
||||
else:
|
||||
has_remote_ocr_action = self.instance is not None and (
|
||||
self.instance.actions.filter(
|
||||
type=WorkflowAction.WorkflowActionType.REMOTE_OCR,
|
||||
).exists()
|
||||
)
|
||||
has_ai_suggestions_action = self.instance is not None and (
|
||||
self.instance.actions.filter(
|
||||
type=WorkflowAction.WorkflowActionType.APPLY_AI_SUGGESTIONS,
|
||||
).exists()
|
||||
)
|
||||
|
||||
if "triggers" in attrs:
|
||||
has_consumption_trigger = any(
|
||||
trigger.get("type") == WorkflowTrigger.WorkflowTriggerType.CONSUMPTION
|
||||
for trigger in attrs["triggers"]
|
||||
)
|
||||
has_non_consumption_trigger = any(
|
||||
trigger.get("type") != WorkflowTrigger.WorkflowTriggerType.CONSUMPTION
|
||||
for trigger in attrs["triggers"]
|
||||
)
|
||||
else:
|
||||
has_consumption_trigger = self.instance is not None and (
|
||||
self.instance.triggers.filter(
|
||||
type=WorkflowTrigger.WorkflowTriggerType.CONSUMPTION,
|
||||
).exists()
|
||||
)
|
||||
has_non_consumption_trigger = self.instance is not None and (
|
||||
self.instance.triggers.exclude(
|
||||
type=WorkflowTrigger.WorkflowTriggerType.CONSUMPTION,
|
||||
).exists()
|
||||
)
|
||||
|
||||
# Remote OCR can only work with consumption triggers
|
||||
if has_remote_ocr_action and not has_consumption_trigger:
|
||||
raise serializers.ValidationError(
|
||||
"Remote OCR actions require a consumption started trigger",
|
||||
)
|
||||
|
||||
# Suggestions are made from the document content, which does not exist
|
||||
# until after consumption has finished
|
||||
if has_ai_suggestions_action and not has_non_consumption_trigger:
|
||||
raise serializers.ValidationError(
|
||||
"Apply AI suggestions actions require a trigger other than "
|
||||
"consumption started",
|
||||
)
|
||||
|
||||
return attrs
|
||||
|
||||
def update_triggers_and_actions(
|
||||
self,
|
||||
instance: Workflow,
|
||||
triggers,
|
||||
actions,
|
||||
) -> None:
|
||||
set_triggers = []
|
||||
set_actions = []
|
||||
|
||||
if triggers is not None and triggers is not serializers.empty:
|
||||
for trigger in triggers:
|
||||
filter_has_tags = trigger.pop("filter_has_tags", None)
|
||||
filter_has_all_tags = trigger.pop("filter_has_all_tags", None)
|
||||
filter_has_not_tags = trigger.pop("filter_has_not_tags", None)
|
||||
filter_has_any_correspondents = trigger.pop(
|
||||
"filter_has_any_correspondents",
|
||||
None,
|
||||
)
|
||||
filter_has_not_correspondents = trigger.pop(
|
||||
"filter_has_not_correspondents",
|
||||
None,
|
||||
)
|
||||
filter_has_any_document_types = trigger.pop(
|
||||
"filter_has_any_document_types",
|
||||
None,
|
||||
)
|
||||
filter_has_not_document_types = trigger.pop(
|
||||
"filter_has_not_document_types",
|
||||
None,
|
||||
)
|
||||
filter_has_any_storage_paths = trigger.pop(
|
||||
"filter_has_any_storage_paths",
|
||||
None,
|
||||
)
|
||||
filter_has_not_storage_paths = trigger.pop(
|
||||
"filter_has_not_storage_paths",
|
||||
None,
|
||||
)
|
||||
# Convert sources to strings to handle django-multiselectfield v1.0 changes
|
||||
WorkflowTriggerSerializer.normalize_workflow_trigger_sources(trigger)
|
||||
trigger_instance, _ = WorkflowTrigger.objects.update_or_create(
|
||||
id=trigger.get("id"),
|
||||
defaults=trigger,
|
||||
)
|
||||
if filter_has_tags is not None:
|
||||
trigger_instance.filter_has_tags.set(filter_has_tags)
|
||||
if filter_has_all_tags is not None:
|
||||
trigger_instance.filter_has_all_tags.set(filter_has_all_tags)
|
||||
if filter_has_not_tags is not None:
|
||||
trigger_instance.filter_has_not_tags.set(filter_has_not_tags)
|
||||
if filter_has_any_correspondents is not None:
|
||||
trigger_instance.filter_has_any_correspondents.set(
|
||||
filter_has_any_correspondents,
|
||||
)
|
||||
if filter_has_not_correspondents is not None:
|
||||
trigger_instance.filter_has_not_correspondents.set(
|
||||
filter_has_not_correspondents,
|
||||
)
|
||||
if filter_has_any_document_types is not None:
|
||||
trigger_instance.filter_has_any_document_types.set(
|
||||
filter_has_any_document_types,
|
||||
)
|
||||
if filter_has_not_document_types is not None:
|
||||
trigger_instance.filter_has_not_document_types.set(
|
||||
filter_has_not_document_types,
|
||||
)
|
||||
if filter_has_any_storage_paths is not None:
|
||||
trigger_instance.filter_has_any_storage_paths.set(
|
||||
filter_has_any_storage_paths,
|
||||
)
|
||||
if filter_has_not_storage_paths is not None:
|
||||
trigger_instance.filter_has_not_storage_paths.set(
|
||||
filter_has_not_storage_paths,
|
||||
)
|
||||
set_triggers.append(trigger_instance)
|
||||
|
||||
if actions is not None and actions is not serializers.empty:
|
||||
for index, action in enumerate(actions):
|
||||
action["order"] = index
|
||||
assign_tags = action.pop("assign_tags", None)
|
||||
assign_view_users = action.pop("assign_view_users", None)
|
||||
assign_view_groups = action.pop("assign_view_groups", None)
|
||||
assign_change_users = action.pop("assign_change_users", None)
|
||||
assign_change_groups = action.pop("assign_change_groups", None)
|
||||
assign_custom_fields = action.pop("assign_custom_fields", None)
|
||||
remove_tags = action.pop("remove_tags", None)
|
||||
remove_correspondents = action.pop("remove_correspondents", None)
|
||||
remove_document_types = action.pop("remove_document_types", None)
|
||||
remove_storage_paths = action.pop("remove_storage_paths", None)
|
||||
remove_custom_fields = action.pop("remove_custom_fields", None)
|
||||
remove_owners = action.pop("remove_owners", None)
|
||||
remove_view_users = action.pop("remove_view_users", None)
|
||||
remove_view_groups = action.pop("remove_view_groups", None)
|
||||
remove_change_users = action.pop("remove_change_users", None)
|
||||
remove_change_groups = action.pop("remove_change_groups", None)
|
||||
|
||||
email_data = action.pop("email", None)
|
||||
webhook_data = action.pop("webhook", None)
|
||||
|
||||
action_instance, _ = WorkflowAction.objects.update_or_create(
|
||||
id=action.get("id"),
|
||||
defaults=action,
|
||||
)
|
||||
|
||||
if email_data is not None:
|
||||
serializer = WorkflowActionEmailSerializer(data=email_data)
|
||||
serializer.is_valid(raise_exception=True)
|
||||
email, _ = WorkflowActionEmail.objects.update_or_create(
|
||||
id=email_data.get("id"),
|
||||
defaults=serializer.validated_data,
|
||||
)
|
||||
action_instance.email = email
|
||||
action_instance.save()
|
||||
|
||||
if webhook_data is not None:
|
||||
serializer = WorkflowActionWebhookSerializer(data=webhook_data)
|
||||
serializer.is_valid(raise_exception=True)
|
||||
webhook, _ = WorkflowActionWebhook.objects.update_or_create(
|
||||
id=webhook_data.get("id"),
|
||||
defaults=serializer.validated_data,
|
||||
)
|
||||
action_instance.webhook = webhook
|
||||
action_instance.save()
|
||||
|
||||
if assign_tags is not None:
|
||||
action_instance.assign_tags.set(assign_tags)
|
||||
if assign_view_users is not None:
|
||||
action_instance.assign_view_users.set(assign_view_users)
|
||||
if assign_view_groups is not None:
|
||||
action_instance.assign_view_groups.set(assign_view_groups)
|
||||
if assign_change_users is not None:
|
||||
action_instance.assign_change_users.set(assign_change_users)
|
||||
if assign_change_groups is not None:
|
||||
action_instance.assign_change_groups.set(assign_change_groups)
|
||||
if assign_custom_fields is not None:
|
||||
action_instance.assign_custom_fields.set(assign_custom_fields)
|
||||
if remove_tags is not None:
|
||||
action_instance.remove_tags.set(remove_tags)
|
||||
if remove_correspondents is not None:
|
||||
action_instance.remove_correspondents.set(remove_correspondents)
|
||||
if remove_document_types is not None:
|
||||
action_instance.remove_document_types.set(remove_document_types)
|
||||
if remove_storage_paths is not None:
|
||||
action_instance.remove_storage_paths.set(remove_storage_paths)
|
||||
if remove_custom_fields is not None:
|
||||
action_instance.remove_custom_fields.set(remove_custom_fields)
|
||||
if remove_owners is not None:
|
||||
action_instance.remove_owners.set(remove_owners)
|
||||
if remove_view_users is not None:
|
||||
action_instance.remove_view_users.set(remove_view_users)
|
||||
if remove_view_groups is not None:
|
||||
action_instance.remove_view_groups.set(remove_view_groups)
|
||||
if remove_change_users is not None:
|
||||
action_instance.remove_change_users.set(remove_change_users)
|
||||
if remove_change_groups is not None:
|
||||
action_instance.remove_change_groups.set(remove_change_groups)
|
||||
|
||||
set_actions.append(action_instance)
|
||||
|
||||
if triggers is not serializers.empty:
|
||||
instance.triggers.set(set_triggers)
|
||||
if actions is not serializers.empty:
|
||||
instance.actions.set(set_actions)
|
||||
instance.save()
|
||||
|
||||
def prune_triggers_and_actions(self) -> None:
|
||||
"""
|
||||
ManyToMany fields dont support e.g. on_delete so we need to discard unattached
|
||||
triggers and actions manually
|
||||
"""
|
||||
WorkflowTrigger.objects.annotate(
|
||||
workflow_count=Count("workflows"),
|
||||
).filter(workflow_count=0).delete()
|
||||
|
||||
WorkflowAction.objects.annotate(
|
||||
workflow_count=Count("workflows"),
|
||||
).filter(workflow_count=0).delete()
|
||||
|
||||
WorkflowActionEmail.objects.filter(action=None).delete()
|
||||
WorkflowActionWebhook.objects.filter(action=None).delete()
|
||||
|
||||
def create(self, validated_data) -> Workflow:
|
||||
if "triggers" in validated_data:
|
||||
triggers = validated_data.pop("triggers")
|
||||
|
||||
if "actions" in validated_data:
|
||||
actions = validated_data.pop("actions")
|
||||
for action in actions:
|
||||
action.pop("id", None)
|
||||
|
||||
instance = super().create(validated_data)
|
||||
|
||||
self.update_triggers_and_actions(instance, triggers, actions)
|
||||
|
||||
return instance
|
||||
|
||||
def update(self, instance: Workflow, validated_data) -> Workflow:
|
||||
triggers = validated_data.pop("triggers", serializers.empty)
|
||||
actions = validated_data.pop("actions", serializers.empty)
|
||||
|
||||
instance = super().update(instance, validated_data)
|
||||
|
||||
self.update_triggers_and_actions(instance, triggers, actions)
|
||||
self.prune_triggers_and_actions()
|
||||
|
||||
return instance
|
||||
@@ -4,10 +4,10 @@ from rest_framework.exceptions import PermissionDenied
|
||||
|
||||
from documents.permissions import get_objects_for_user_owner_aware
|
||||
from documents.permissions import has_perms_owner_aware
|
||||
from documents.serialisers import CorrespondentField
|
||||
from documents.serialisers import DocumentTypeField
|
||||
from documents.serialisers import OwnedObjectSerializer
|
||||
from documents.serialisers import TagsField
|
||||
from documents.serialisers.base import OwnedObjectSerializer
|
||||
from documents.serialisers.metadata import CorrespondentField
|
||||
from documents.serialisers.metadata import DocumentTypeField
|
||||
from documents.serialisers.metadata import TagsField
|
||||
from paperless_mail.models import MailAccount
|
||||
from paperless_mail.models import MailRule
|
||||
from paperless_mail.models import ProcessedMail
|
||||
|
||||
Reference in New Issue
Block a user