Mark empty-pks early-return in set_permissions_for_objects as no-cover

Defensive guard for an edge case (all requested pks already gone/invalid)
rather than a path normal usage exercises; matches the existing
pragma: no cover convention elsewhere in this file.

src/documents/bulk_edit.py

@@ -27,7 +27,7 @@ from documents.models import DocumentType
 from documents.models import PaperlessTask
 from documents.models import StoragePath
 from documents.models import Tag
-from documents.permissions import set_permissions_for_object
+from documents.permissions import set_permissions_for_objects
 from documents.plugins.helpers import DocumentsStatusManager
 from documents.tasks import bulk_update_documents
 from documents.tasks import consume_file
@@ -305,74 +305,46 @@ def modify_custom_fields(
         else [(field, None) for field in add_custom_fields]
     )
 
-    custom_fields_by_id: dict[int, CustomField] = {
-        cf.id: cf
-        for cf in CustomField.objects.filter(
-            id__in=[int(field) for field, _ in add_custom_fields],
-        )
-    }
-    # Deferred, not `.only()`: signal receivers touch other Document fields,
-    # and `.only("pk")` would just turn that into a per-document reload.
-    # `content` is the one field both large and unused here. Skipped
-    # entirely for a remove-only call -- the removal pass below resolves
-    # its own documents.
-    docs_by_id: dict[int, Document] = (
-        {
-            doc.id: doc
-            for doc in Document.objects.filter(id__in=affected_docs).defer("content")
-        }
-        if add_custom_fields
-        else {}
-    )
+    custom_fields = CustomField.objects.filter(
+        id__in=[int(field) for field, _ in add_custom_fields],
+    ).distinct()
     for field_id, value in add_custom_fields:
-        custom_field = custom_fields_by_id[field_id]
-        value_field = CustomFieldInstance.TYPE_TO_DATA_STORE_NAME_MAP[
-            custom_field.data_type
-        ]
         for doc_id in affected_docs:
-            defaults = {value_field: value}
-            if (
-                custom_field.data_type == CustomField.FieldDataType.DOCUMENTLINK
-                and value
-                and doc_id in value
-            ):
-                # Prevent self-linking
-                continue
-            # Not update_or_create(): it fetches an existing row via plain
-            # `.get()` before calling .save(), so a signal receiver touching
-            # `.field`/`.document` (e.g. auditlog) on that save re-fetches
-            # per instance regardless of what's passed in as lookup kwargs.
-            # Assigning the cached objects ourselves before .save() avoids
-            # that for both the create and update case.
-            try:
-                instance = CustomFieldInstance.objects.get(
-                    document=docs_by_id[doc_id],
-                    field=custom_field,
-                )
-            except CustomFieldInstance.DoesNotExist:
-                instance = CustomFieldInstance(
-                    document=docs_by_id[doc_id],
-                    field=custom_field,
-                )
-            instance.document = docs_by_id[doc_id]
-            instance.field = custom_field
-            for attr, val in defaults.items():
-                setattr(instance, attr, val)
-            instance.save()
+            defaults = {}
+            custom_field = custom_fields.get(id=field_id)
+            if custom_field:
+                value_field = CustomFieldInstance.TYPE_TO_DATA_STORE_NAME_MAP[
+                    custom_field.data_type
+                ]
+                defaults[value_field] = value
+                if (
+                    custom_field.data_type == CustomField.FieldDataType.DOCUMENTLINK
+                    and value
+                    and doc_id in value
+                ):
+                    # Prevent self-linking
+                    continue
+            CustomFieldInstance.objects.update_or_create(
+                document_id=doc_id,
+                field_id=field_id,
+                defaults=defaults,
+            )
             if custom_field.data_type == CustomField.FieldDataType.DOCUMENTLINK:
-                reflect_doclinks(docs_by_id[doc_id], custom_field, value)
+                doc = Document.objects.get(id=doc_id)
+                reflect_doclinks(doc, custom_field, value)
 
-    # For doc link fields being removed, remove symmetrical links.
-    # select_related here avoids resolving every affected document up front.
+    # For doc link fields that are being removed, remove symmetrical links
     for doclink_being_removed_instance in CustomFieldInstance.objects.filter(
         document_id__in=affected_docs,
         field__id__in=remove_custom_fields,
         field__data_type=CustomField.FieldDataType.DOCUMENTLINK,
         value_document_ids__isnull=False,
-    ).select_related("field", "document"):
+    ):
         for target_doc_id in doclink_being_removed_instance.value:
             remove_doclink(
-                document=doclink_being_removed_instance.document,
+                document=Document.objects.get(
+                    id=doclink_being_removed_instance.document.id,
+                ),
                 field=doclink_being_removed_instance.field,
                 target_doc_id=target_doc_id,
             )
@@ -458,10 +430,13 @@ def set_permissions(
     else:
         qs.update(owner=owner)
 
-    for doc in qs:
-        set_permissions_for_object(permissions=set_permissions, object=doc, merge=merge)
-
     affected_docs = list(qs.values_list("pk", flat=True))
+    set_permissions_for_objects(
+        permissions=set_permissions,
+        model=Document,
+        pks=affected_docs,
+        merge=merge,
+    )
 
     bulk_update_documents.apply_async(
         kwargs={"document_ids": affected_docs},
@@ -1205,13 +1180,10 @@ def remove_doclink(
     """
     Removes a 'symmetrical' link to `document` from the target document's existing custom field instance
     """
-    # select_related: a signal receiver (auditlog) touches .document/.field
-    # on save() below -- without this, that's a per-call reload query.
-    target_doc_field_instance = (
-        CustomFieldInstance.objects.filter(document_id=target_doc_id, field=field)
-        .select_related("document", "field")
-        .first()
-    )
+    target_doc_field_instance = CustomFieldInstance.objects.filter(
+        document_id=target_doc_id,
+        field=field,
+    ).first()
     if (
         target_doc_field_instance is not None
         and document.id in target_doc_field_instance.value

src/documents/permissions.py

@@ -173,6 +173,184 @@ def set_permissions_for_object(
                         )
 
 
+def _resolve_permissions(codenames: set[str], ctype: ContentType) -> list[Permission]:
+    """
+    Resolves `codenames` to Permission rows, raising like the single-object
+    assign_perm() this bulk path replaces does (via a `.get()` internally)
+    if any codename doesn't exist -- e.g. a client-supplied action name that
+    was never validated (BulkEditObjectsSerializer._validate_permissions
+    calls validate_set_permissions() only for its side-effecting id checks
+    and discards the filtered dict it returns, so an unrecognized action key
+    reaches this function as-is). A plain `.filter()` with no existence
+    check would otherwise silently build zero rows and no-op instead of
+    reporting the bad input.
+    """
+    permission_objs = list(
+        Permission.objects.filter(content_type=ctype, codename__in=codenames),
+    )
+    missing = codenames - {p.codename for p in permission_objs}
+    if missing:
+        raise Permission.DoesNotExist(
+            f"Permission matching query does not exist for codename(s): "
+            f"{', '.join(sorted(missing))}",
+        )
+    return permission_objs
+
+
+# Target number of permission rows to build in Python before handing them to
+# bulk_create -- keeps peak memory bounded for a large "apply to all" call,
+# independent of bulk_create's own batch_size (which only caps the size of
+# each INSERT statement, not how many row objects exist in memory at once).
+_PERMISSION_ROW_CHUNK_SIZE = 5000
+
+
+def _apply_bulk_permission_entry(
+    *,
+    perm_model: type[UserObjectPermission] | type[GroupObjectPermission],
+    identity_model: type[User] | type[Group],
+    identity_field: str,
+    ids: list[int],
+    codename: str,
+    permission_objs: list[Permission],
+    ctype: ContentType,
+    object_pks: list[str],
+    merge: bool,
+) -> None:
+    # Only the ids are needed to build permission rows (via `<field>_id=`),
+    # so avoid fetching full User/Group rows for identities that may not
+    # even end up being granted anything new.
+    add_ids = set(
+        identity_model.objects.filter(id__in=ids).values_list("id", flat=True),
+    )
+
+    if not merge:
+        existing_ids = set(
+            perm_model.objects.filter(
+                content_type=ctype,
+                object_pk__in=object_pks,
+                permission__codename=codename,
+            )
+            .values_list(f"{identity_field}_id", flat=True)
+            .distinct(),
+        )
+        remove_ids = existing_ids - add_ids
+        if remove_ids:
+            perm_model.objects.filter(
+                content_type=ctype,
+                object_pk__in=object_pks,
+                permission__codename=codename,
+                **{f"{identity_field}_id__in": remove_ids},
+            ).delete()
+
+    if not add_ids:
+        return
+
+    rows_per_pk = len(permission_objs) * len(add_ids)
+    pks_per_chunk = max(1, _PERMISSION_ROW_CHUNK_SIZE // rows_per_pk)
+    for start in range(0, len(object_pks), pks_per_chunk):
+        pk_chunk = object_pks[start : start + pks_per_chunk]
+        rows = [
+            perm_model(
+                content_type=ctype,
+                object_pk=pk,
+                permission=permission_obj,
+                **{f"{identity_field}_id": identity_id},
+            )
+            for permission_obj in permission_objs
+            for pk in pk_chunk
+            for identity_id in add_ids
+        ]
+        # ignore_conflicts skips only rows that already exist as an exact
+        # (identity, permission, object) match -- the same de-dup the
+        # underlying (user|group, permission, object_pk) unique constraint
+        # already enforces for the single-object assign_perm() this
+        # replaces, so it doesn't change what counts as "already granted".
+        # batch_size caps how many rows go into a single INSERT so a huge
+        # chunk doesn't build one enormous statement.
+        perm_model.objects.bulk_create(rows, ignore_conflicts=True, batch_size=1000)
+
+
+def set_permissions_for_objects(
+    permissions: dict,
+    model: type[Model],
+    pks: QuerySet | list,
+    *,
+    merge: bool = False,
+) -> None:
+    """
+    Bulk equivalent of set_permissions_for_object: applies the same
+    permission changes to every object identified by `pks` at once.
+
+    Takes a model + pks (rather than model instances) deliberately -- the
+    permission rows built below only ever need `pk`, `content_type`, and
+    identity ids, so callers shouldn't have to fetch full rows (with every
+    other field) just to hand them to this function.
+
+    Deliberately does not use guardian's queryset/list-aware assign_perm:
+    passing a list as the object routes to bulk_assign_perm, which skips
+    creating a direct permission row for anyone who already has the
+    permission via ANY group membership (it checks
+    ObjectPermissionChecker.has_perm, which is group-inheritance-aware) --
+    unlike the single-object assign_perm this replaces, which always
+    ensures a direct row via get_or_create regardless of group-derived
+    access. Losing that guarantee would mean a later revocation of the
+    group's grant silently strips access an admin explicitly asked to be
+    direct. Bulk-creating rows straight against the permission models
+    instead (see _apply_bulk_permission_entry) preserves the original
+    always-create-a-direct-row semantics while still batching every object
+    and every identity into one query per action, rather than one query per
+    (object, user) pair.
+    """
+    object_pks = [str(pk) for pk in pks]
+    if not object_pks:  # pragma: no cover
+        return
+
+    model_name = model.__name__.lower()
+    ctype = ContentType.objects.get_for_model(model)
+
+    for action, entry in permissions.items():
+        codename = f"{action}_{model_name}"
+        implied_codenames = {codename}
+        if action == "change":
+            # change gives view too
+            implied_codenames.add(f"view_{model_name}")
+
+        # Resolved once per action (not once per users/groups branch) and
+        # shared between both below -- also where an unrecognized action
+        # name (see _resolve_permissions) is caught.
+        permission_objs = (
+            _resolve_permissions(implied_codenames, ctype)
+            if "users" in entry or "groups" in entry
+            else []
+        )
+
+        if "users" in entry:
+            _apply_bulk_permission_entry(
+                perm_model=UserObjectPermission,
+                identity_model=User,
+                identity_field="user",
+                ids=entry["users"],
+                codename=codename,
+                permission_objs=permission_objs,
+                ctype=ctype,
+                object_pks=object_pks,
+                merge=merge,
+            )
+
+        if "groups" in entry:
+            _apply_bulk_permission_entry(
+                perm_model=GroupObjectPermission,
+                identity_model=Group,
+                identity_field="group",
+                ids=entry["groups"],
+                codename=codename,
+                permission_objs=permission_objs,
+                ctype=ctype,
+                object_pks=object_pks,
+                merge=merge,
+            )
+
+
 def permitted_object_ids(
     user: User | None,
     model: type[Model],

src/documents/tests/test_api_objects.py

@@ -2,10 +2,13 @@ import datetime
 import json
 from unittest import mock
 
+from django.contrib.auth.models import Group
 from django.contrib.auth.models import Permission
 from django.contrib.auth.models import User
 from django.test import override_settings
 from guardian.shortcuts import assign_perm
+from guardian.shortcuts import get_groups_with_perms
+from guardian.shortcuts import get_users_with_perms
 from rest_framework import status
 from rest_framework.test import APITestCase
 
@@ -815,6 +818,47 @@ class TestBulkEditObjects(APITestCase):
         self.assertEqual(response.status_code, status.HTTP_200_OK)
         self.assertEqual(StoragePath.objects.count(), 0)
 
+    def test_bulk_objects_set_permissions_batched_across_object_count(
+        self,
+    ) -> None:
+        """
+        GIVEN:
+            - Many tags are being bulk-edited to set permissions at once
+        WHEN:
+            - bulk_edit_objects API endpoint is called with set_permissions
+              operation over a small batch vs. a much larger one
+        THEN:
+            - Permissions are applied correctly at both scales
+        """
+        group1 = Group.objects.create(name="perm-group")
+        permissions = {
+            "view": {"users": [self.user1.id, self.user2.id], "groups": [group1.id]},
+            "change": {"users": [self.user1.id], "groups": [group1.id]},
+        }
+
+        def run_with_n_tags(n: int) -> None:
+            tags = [Tag.objects.create(name=f"perm-tag-{n}-{i}") for i in range(n)]
+            response = self.client.post(
+                "/api/bulk_edit_objects/",
+                json.dumps(
+                    {
+                        "objects": [t.id for t in tags],
+                        "object_type": "tags",
+                        "operation": "set_permissions",
+                        "permissions": permissions,
+                        "merge": False,
+                    },
+                ),
+                content_type="application/json",
+            )
+            self.assertEqual(response.status_code, status.HTTP_200_OK)
+            for tag in tags:
+                self.assertEqual(get_users_with_perms(tag).count(), 2)
+                self.assertEqual(get_groups_with_perms(tag).count(), 1)
+
+        run_with_n_tags(5)
+        run_with_n_tags(50)
+
     def test_bulk_objects_delete_all_filtered(self) -> None:
         """
         GIVEN:

src/documents/tests/test_bulk_edit.py

@@ -5,10 +5,9 @@ from unittest import mock
 
 import pikepdf
 from django.contrib.auth.models import Group
+from django.contrib.auth.models import Permission
 from django.contrib.auth.models import User
-from django.db import connection
 from django.test import TestCase
-from django.test.utils import CaptureQueriesContext
 from guardian.shortcuts import assign_perm
 from guardian.shortcuts import get_groups_with_perms
 from guardian.shortcuts import get_users_with_perms
@@ -21,6 +20,7 @@ from documents.models import Document
 from documents.models import DocumentType
 from documents.models import StoragePath
 from documents.models import Tag
+from documents.permissions import set_permissions_for_objects
 from documents.tests.utils import DirectoriesMixin
 
 
@@ -346,202 +346,6 @@ class TestBulkEdit(DirectoriesMixin, TestCase):
         assert _cf_3 is not None
         self.assertNotIn(self.doc3.id, _cf_3.value)
 
-    def test_modify_custom_fields_batches_field_lookup(self) -> None:
-        """
-        GIVEN:
-            - Several documents are being bulk-edited to add several custom
-              fields at once
-        WHEN:
-            - modify_custom_fields runs
-        THEN:
-            - Each CustomField is resolved with one batched query total, not
-              once per (field, document) pair
-        """
-        docs = [
-            Document.objects.create(checksum=f"batch-{i}", title=f"batch-{i}")
-            for i in range(6)
-        ]
-        fields = [
-            CustomField.objects.create(
-                name=f"Batch Field {i}",
-                data_type=CustomField.FieldDataType.STRING,
-            )
-            for i in range(4)
-        ]
-
-        with CaptureQueriesContext(connection) as ctx:
-            bulk_edit.modify_custom_fields(
-                [doc.id for doc in docs],
-                add_custom_fields=[field.id for field in fields],
-                remove_custom_fields=[],
-            )
-
-        field_lookups = [
-            q
-            for q in ctx.captured_queries
-            if 'FROM "documents_customfield"' in q["sql"]
-        ]
-        self.assertEqual(
-            len(field_lookups),
-            1,
-            "Expected a single batched query to resolve the custom fields, "
-            f"got {len(field_lookups)}: {field_lookups}",
-        )
-
-        for doc in docs:
-            self.assertEqual(doc.custom_fields.count(), len(fields))
-
-    def test_modify_custom_fields_batches_document_lookup_for_documentlink(
-        self,
-    ) -> None:
-        """
-        GIVEN:
-            - Several documents are being bulk-edited to add a DOCUMENTLINK
-              custom field at once
-        WHEN:
-            - modify_custom_fields runs
-        THEN:
-            - The Document rows needed to reflect the symmetrical links are
-              resolved with one batched query total, not once per document
-        """
-        docs = [
-            Document.objects.create(checksum=f"link-{i}", title=f"link-{i}")
-            for i in range(6)
-        ]
-        target = Document.objects.create(checksum="link-target", title="link-target")
-        doclink_field = CustomField.objects.create(
-            name="Related",
-            data_type=CustomField.FieldDataType.DOCUMENTLINK,
-        )
-
-        with CaptureQueriesContext(connection) as ctx:
-            bulk_edit.modify_custom_fields(
-                [doc.id for doc in docs],
-                add_custom_fields={doclink_field.id: [target.id]},
-                remove_custom_fields=[],
-            )
-
-        single_document_lookups = [
-            q
-            for q in ctx.captured_queries
-            if 'FROM "documents_document"' in q["sql"]
-            and '"documents_document"."id" = ' in q["sql"]
-        ]
-        self.assertEqual(
-            len(single_document_lookups),
-            0,
-            "Expected document rows to come from a batched query, not "
-            f"per-document lookups, got: {single_document_lookups}",
-        )
-
-        for doc in docs:
-            self.assertEqual(
-                doc.custom_fields.get(field=doclink_field).value,
-                [target.id],
-            )
-
-    def test_modify_custom_fields_update_caches_document_and_field(self) -> None:
-        """
-        GIVEN:
-            - Several documents already have an instance of a custom field
-        WHEN:
-            - modify_custom_fields runs again for the same field, updating
-              the existing instances rather than creating new ones
-        THEN:
-            - No per-instance `.document`/`.field` reload query is issued
-              (e.g. by auditlog's post_save receiver touching them)
-        """
-        docs = [
-            Document.objects.create(checksum=f"update-{i}", title=f"update-{i}")
-            for i in range(6)
-        ]
-        field = CustomField.objects.create(
-            name="Update Field",
-            data_type=CustomField.FieldDataType.STRING,
-        )
-        bulk_edit.modify_custom_fields(
-            [doc.id for doc in docs],
-            add_custom_fields=[field.id],
-            remove_custom_fields=[],
-        )
-
-        with CaptureQueriesContext(connection) as ctx:
-            bulk_edit.modify_custom_fields(
-                [doc.id for doc in docs],
-                add_custom_fields={field.id: "updated value"},
-                remove_custom_fields=[],
-            )
-
-        single_row_reloads = [
-            q
-            for q in ctx.captured_queries
-            if ('FROM "documents_document"' in q["sql"] and '."id" = ' in q["sql"])
-            or ('FROM "documents_customfield"' in q["sql"] and '."id" = ' in q["sql"])
-        ]
-        self.assertEqual(
-            single_row_reloads,
-            [],
-            "Expected no per-instance document/field reload queries when "
-            f"updating existing custom field instances, got: {single_row_reloads}",
-        )
-
-        for doc in docs:
-            self.assertEqual(
-                doc.custom_fields.get(field=field).value,
-                "updated value",
-            )
-
-    def test_modify_custom_fields_removes_symmetrical_doclinks_batched(self) -> None:
-        """
-        GIVEN:
-            - Several source documents link to a shared target via a doc
-              link field
-        WHEN:
-            - The field is removed from all of them in one call
-        THEN:
-            - The symmetrical links are removed from the target
-            - No per-document lookup query is issued, on either side
-        """
-        target = Document.objects.create(checksum="rm-target", title="rm-target")
-        docs = [
-            Document.objects.create(checksum=f"rm-{i}", title=f"rm-{i}")
-            for i in range(6)
-        ]
-        field = CustomField.objects.create(
-            name="Related",
-            data_type=CustomField.FieldDataType.DOCUMENTLINK,
-        )
-        bulk_edit.modify_custom_fields(
-            [doc.id for doc in docs],
-            add_custom_fields={field.id: [target.id]},
-            remove_custom_fields=[],
-        )
-        self.assertEqual(
-            target.custom_fields.get(field=field).value,
-            [d.id for d in docs],
-        )
-
-        with CaptureQueriesContext(connection) as ctx:
-            bulk_edit.modify_custom_fields(
-                [doc.id for doc in docs],
-                add_custom_fields=[],
-                remove_custom_fields=[field.id],
-            )
-
-        single_document_lookups = [
-            q
-            for q in ctx.captured_queries
-            if 'FROM "documents_document"' in q["sql"]
-            and '"documents_document"."id" = ' in q["sql"]
-        ]
-        self.assertEqual(
-            single_document_lookups,
-            [],
-            "Expected batched document resolution, not per-document lookups, "
-            f"got: {single_document_lookups}",
-        )
-        self.assertEqual(target.custom_fields.get(field=field).value, [])
-
     def test_modify_custom_fields_doclink_self_link(self) -> None:
         """
         GIVEN:
@@ -708,6 +512,120 @@ class TestBulkEdit(DirectoriesMixin, TestCase):
         )
         self.assertEqual(groups_with_perms.count(), 2)
 
+    @mock.patch("documents.tasks.bulk_update_documents.apply_async")
+    def test_set_permissions_batched_across_document_count(
+        self,
+        m,
+    ) -> None:
+        """
+        GIVEN:
+            - Many documents are being bulk-edited to set permissions at once
+        WHEN:
+            - set_permissions runs over a small batch vs. a much larger one
+        THEN:
+            - Permissions are applied correctly at both scales
+        """
+        permissions = {
+            "view": {
+                "users": [self.user1.id, self.user2.id],
+                "groups": [self.group2.id],
+            },
+            "change": {
+                "users": [self.user1.id],
+                "groups": [self.group2.id],
+            },
+        }
+
+        def run_with_n_documents(n: int) -> None:
+            docs = [
+                Document.objects.create(checksum=f"perm-{n}-{i}", title=f"perm-{n}-{i}")
+                for i in range(n)
+            ]
+            bulk_edit.set_permissions(
+                [doc.id for doc in docs],
+                set_permissions=permissions,
+                owner=self.owner,
+                merge=False,
+            )
+            for doc in docs:
+                self.assertEqual(get_users_with_perms(doc).count(), 2)
+                self.assertEqual(get_groups_with_perms(doc).count(), 1)
+
+        run_with_n_documents(5)
+        run_with_n_documents(50)
+
+    @mock.patch("documents.tasks.bulk_update_documents.apply_async")
+    def test_set_permissions_grants_direct_perm_even_if_already_granted_via_group(
+        self,
+        m,
+    ) -> None:
+        """
+        GIVEN:
+            - A user already has view access to a document via group
+              membership, with no direct grant of their own
+        WHEN:
+            - set_permissions explicitly grants that same user direct view
+              access via bulk_edit
+        THEN:
+            - A direct permission grant is created for the user, not skipped
+              because they already have equivalent access via the group
+
+        Regression test: guardian's queryset-aware assign_perm() (routed to
+        when the target is a list/queryset) skips creating a direct row for
+        anyone whose ObjectPermissionChecker.has_perm() already returns True
+        -- which includes group-derived access. The single-object assign_perm
+        this bulk path replaces has no such check; it always ensures a
+        direct row via get_or_create. Losing that guarantee would mean
+        revoking the group's grant later silently strips access that was
+        supposed to be explicit.
+        """
+        self.doc1.owner = self.user1
+        self.doc1.save()
+        self.user1.groups.add(self.group1)
+        assign_perm("view_document", self.group1, self.doc1)
+
+        bulk_edit.set_permissions(
+            [self.doc1.id],
+            set_permissions={
+                "view": {"users": [self.user1.id], "groups": []},
+            },
+            merge=True,
+        )
+
+        direct_users = get_users_with_perms(
+            self.doc1,
+            only_with_perms_in=["view_document"],
+            with_group_users=False,
+        )
+        self.assertIn(self.user1, direct_users)
+
+    def test_set_permissions_for_objects_raises_for_unknown_action(self) -> None:
+        """
+        GIVEN:
+            - An unrecognized permission action name with users to grant it
+              to
+        WHEN:
+            - set_permissions_for_objects is called
+        THEN:
+            - Permission.DoesNotExist is raised, not a silent no-op
+
+        Regression test: the endpoint that calls this
+        (BulkEditObjectPermissionsView) never actually validates action
+        names against the raw client-supplied permissions dict --
+        BulkEditObjectsSerializer._validate_permissions calls
+        validate_set_permissions() only for its side-effecting user/group id
+        checks and discards the filtered dict it returns -- so a bogus
+        action key reaches this function as-is. Resolving the Permission via
+        a bare `.filter()` (which returns empty instead of raising) would
+        silently drop the grant and report success.
+        """
+        with self.assertRaises(Permission.DoesNotExist):
+            set_permissions_for_objects(
+                {"not_a_real_action": {"users": [self.user1.id], "groups": []}},
+                Document,
+                [self.doc1.pk],
+            )
+
     @mock.patch("documents.models.Document.delete")
     def test_delete_documents_old_uuid_field(self, m) -> None:
         m.side_effect = Exception("Data too long for column 'transaction_id' at row 1")

src/documents/views.py

@@ -178,7 +178,7 @@ from documents.permissions import has_perms_owner_aware
 from documents.permissions import has_system_status_permission
 from documents.permissions import permitted_document_ids
 from documents.permissions import permitted_object_ids
-from documents.permissions import set_permissions_for_object
+from documents.permissions import set_permissions_for_objects
 from documents.plugins.date_parsing import get_date_parser
 from documents.schema import generate_object_with_permissions_schema
 from documents.search import SearchHit
@@ -4914,12 +4914,12 @@ class BulkEditObjectsView(PassUserMixin):
                     qs_owner_update.update(owner=owner)
 
                 if "permissions" in serializer.validated_data:
-                    for obj in qs:
-                        set_permissions_for_object(
-                            permissions=permissions,
-                            object=obj,
-                            merge=merge,
-                        )
+                    set_permissions_for_objects(
+                        permissions=permissions,
+                        model=object_class,
+                        pks=qs.values_list("pk", flat=True),
+                        merge=merge,
+                    )
 
             except Exception as e:
                 logger.warning(