mirror of
https://github.com/paperless-ngx/paperless-ngx.git
synced 2026-10-15 04:27:14 +00:00
Chore: type the set-permissions dict
Callers passed whatever they had, including querysets of User and Group objects that only worked because id__in accepts a subquery.
This commit is contained in:
1 parent
e3b4cf55f0
commit
c45c7c7a99
6 files changed
+50
-21
No files matched your search
@@ -34,6 +34,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 PermissionsSpec
|
||||
from documents.permissions import set_permissions_for_objects
|
||||
from documents.plugins.helpers import DocumentsStatusManager
|
||||
from documents.tasks import bulk_update_documents
|
||||
@@ -445,7 +446,7 @@ def reprocess(doc_ids: list[int], *, remote_ocr: bool = False) -> Literal["OK"]:
|
||||
|
||||
def set_permissions(
|
||||
doc_ids: list[int],
|
||||
set_permissions: dict,
|
||||
set_permissions: PermissionsSpec,
|
||||
*,
|
||||
owner: User | None = None,
|
||||
merge: bool = False,
|
||||
|
||||
@@ -38,6 +38,7 @@ from documents.models import StoragePath
|
||||
from documents.models import Tag
|
||||
from documents.models import WorkflowTrigger
|
||||
from documents.parsers import ParseError
|
||||
from documents.permissions import PermissionsSpec
|
||||
from documents.permissions import set_permissions_for_objects
|
||||
from documents.plugins.base import AlwaysRunPluginMixin
|
||||
from documents.plugins.base import ConsumeTaskPlugin
|
||||
@@ -948,7 +949,7 @@ class ConsumerPlugin(
|
||||
or self.metadata.change_users is not None
|
||||
or self.metadata.change_groups is not None
|
||||
):
|
||||
permissions = {
|
||||
permissions: PermissionsSpec = {
|
||||
"view": {
|
||||
"users": self.metadata.view_users or [],
|
||||
"groups": self.metadata.view_groups or [],
|
||||
|
||||
@@ -1,4 +1,10 @@
|
||||
from collections.abc import Iterable
|
||||
from collections.abc import Mapping
|
||||
from typing import Any
|
||||
from typing import Final
|
||||
from typing import Literal
|
||||
from typing import TypeAlias
|
||||
from typing import TypedDict
|
||||
from typing import TypeVar
|
||||
|
||||
from django.contrib.auth.models import Group
|
||||
@@ -27,6 +33,22 @@ from rest_framework.permissions import DjangoObjectPermissions
|
||||
from documents.models import Document
|
||||
from documents.versioning import get_root_document
|
||||
|
||||
PermissionAction = Literal["view", "change"]
|
||||
PERMISSION_ACTIONS: Final[tuple[PermissionAction, ...]] = ("view", "change")
|
||||
|
||||
|
||||
class PermissionGrantees(TypedDict, total=False):
|
||||
"""
|
||||
The user and group ids to grant one permission action to. A missing key
|
||||
leaves that kind of grant untouched.
|
||||
"""
|
||||
|
||||
users: Iterable[int]
|
||||
groups: Iterable[int]
|
||||
|
||||
|
||||
PermissionsSpec: TypeAlias = Mapping[PermissionAction, PermissionGrantees]
|
||||
|
||||
|
||||
class PaperlessObjectPermissions(DjangoObjectPermissions):
|
||||
"""
|
||||
@@ -123,7 +145,7 @@ def _apply_bulk_permission_entry(
|
||||
perm_model: type[UserObjectPermission] | type[GroupObjectPermission],
|
||||
identity_model: type[User] | type[Group],
|
||||
identity_field: str,
|
||||
ids: list[int],
|
||||
ids: Iterable[int],
|
||||
codename: str,
|
||||
permission_objs: list[Permission],
|
||||
ctype: ContentType,
|
||||
@@ -172,7 +194,7 @@ def _apply_bulk_permission_entry(
|
||||
|
||||
|
||||
def set_permissions_for_objects(
|
||||
permissions: dict,
|
||||
permissions: PermissionsSpec,
|
||||
model: type[Model],
|
||||
pks: QuerySet | list,
|
||||
*,
|
||||
@@ -197,7 +219,7 @@ def set_permissions_for_objects(
|
||||
|
||||
# Resolve every action before writing, so a bad one can't leave the
|
||||
# actions ahead of it applied.
|
||||
permissions_by_action: dict[str, list[Permission]] = {}
|
||||
permissions_by_action: dict[PermissionAction, list[Permission]] = {}
|
||||
for action, entry in permissions.items():
|
||||
if "users" not in entry and "groups" not in entry:
|
||||
continue
|
||||
|
||||
@@ -78,6 +78,10 @@ from documents.models import WorkflowActionEmail
|
||||
from documents.models import WorkflowActionWebhook
|
||||
from documents.models import WorkflowTrigger
|
||||
from documents.parsers import is_mime_type_supported
|
||||
from documents.permissions import PERMISSION_ACTIONS
|
||||
from documents.permissions import PermissionAction
|
||||
from documents.permissions import PermissionGrantees
|
||||
from documents.permissions import PermissionsSpec
|
||||
from documents.permissions import get_document_count_filter_for_user
|
||||
from documents.permissions import get_groups_with_only_permission
|
||||
from documents.permissions import has_perms_owner_aware
|
||||
@@ -211,9 +215,6 @@ class MatchingModelSerializer(serializers.ModelSerializer[Any]):
|
||||
return match
|
||||
|
||||
|
||||
PERMISSION_ACTIONS = ("view", "change")
|
||||
|
||||
|
||||
class SetPermissionsMixin:
|
||||
def _validate_user_ids(self, user_ids):
|
||||
users = User.objects.none()
|
||||
@@ -235,8 +236,10 @@ class SetPermissionsMixin:
|
||||
)
|
||||
return groups
|
||||
|
||||
def validate_set_permissions(self, set_permissions=None):
|
||||
permissions_dict = {action: {} for action in PERMISSION_ACTIONS}
|
||||
def validate_set_permissions(self, set_permissions=None) -> PermissionsSpec:
|
||||
permissions_dict: dict[PermissionAction, PermissionGrantees] = {
|
||||
action: {} for action in PERMISSION_ACTIONS
|
||||
}
|
||||
if set_permissions is not None:
|
||||
for action in PERMISSION_ACTIONS:
|
||||
if action in set_permissions:
|
||||
@@ -244,17 +247,17 @@ class SetPermissionsMixin:
|
||||
users = set_permissions[action]["users"]
|
||||
permissions_dict[action]["users"] = self._validate_user_ids(
|
||||
users,
|
||||
)
|
||||
).values_list("id", flat=True)
|
||||
if "groups" in set_permissions[action]:
|
||||
groups = set_permissions[action]["groups"]
|
||||
permissions_dict[action]["groups"] = self._validate_group_ids(
|
||||
groups,
|
||||
)
|
||||
).values_list("id", flat=True)
|
||||
else:
|
||||
del permissions_dict[action]
|
||||
return permissions_dict
|
||||
|
||||
def _set_permissions(self, permissions, object) -> None:
|
||||
def _set_permissions(self, permissions: PermissionsSpec, object) -> None:
|
||||
set_permissions_for_objects(permissions, type(object), [object.pk])
|
||||
|
||||
|
||||
@@ -3086,7 +3089,7 @@ class BulkEditObjectsSerializer(SerializerWithPerms, SetPermissionsMixin):
|
||||
)
|
||||
return objects
|
||||
|
||||
def _validate_permissions(self, permissions) -> dict:
|
||||
def _validate_permissions(self, permissions) -> PermissionsSpec:
|
||||
return self.validate_set_permissions(
|
||||
permissions,
|
||||
)
|
||||
|
||||
@@ -23,6 +23,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 PermissionsSpec
|
||||
from documents.permissions import set_permissions_for_objects
|
||||
from paperless_testing.dirs import DirectoriesMixin
|
||||
from paperless_testing.factories import DocumentFactory
|
||||
@@ -445,7 +446,7 @@ class TestBulkEdit(DirectoriesMixin, TestCase):
|
||||
|
||||
grant_object(self.group1, self.doc1, "view_document")
|
||||
|
||||
permissions = {
|
||||
permissions: PermissionsSpec = {
|
||||
"view": {
|
||||
"users": [self.user1.id, self.user2.id],
|
||||
"groups": [self.group2.id],
|
||||
@@ -488,7 +489,7 @@ class TestBulkEdit(DirectoriesMixin, TestCase):
|
||||
grant_object(self.user1, self.doc1, "view_document")
|
||||
grant_object(self.group1, self.doc1, "view_document")
|
||||
|
||||
permissions = {
|
||||
permissions: PermissionsSpec = {
|
||||
"view": {
|
||||
"users": [self.user2.id],
|
||||
"groups": [self.group2.id],
|
||||
@@ -539,7 +540,7 @@ class TestBulkEdit(DirectoriesMixin, TestCase):
|
||||
batched call rather than one call per (document, identity)
|
||||
pair
|
||||
"""
|
||||
permissions = {
|
||||
permissions: PermissionsSpec = {
|
||||
"view": {
|
||||
"users": [self.user1.id, self.user2.id],
|
||||
"groups": [self.group2.id],
|
||||
@@ -646,7 +647,7 @@ class TestBulkEdit(DirectoriesMixin, TestCase):
|
||||
"""
|
||||
with self.assertRaises(Permission.DoesNotExist):
|
||||
set_permissions_for_objects(
|
||||
{"not_a_real_action": {"users": [self.user1.id], "groups": []}},
|
||||
{"not_a_real_action": {"users": [self.user1.id], "groups": []}}, # type: ignore[bad-assignment]
|
||||
Document,
|
||||
[self.doc1.pk],
|
||||
)
|
||||
@@ -674,7 +675,7 @@ class TestBulkEdit(DirectoriesMixin, TestCase):
|
||||
set_permissions_for_objects(
|
||||
{
|
||||
"view": {"users": [self.user1.id], "groups": []},
|
||||
"not_a_real_action": {"users": [self.user1.id], "groups": []},
|
||||
"not_a_real_action": {"users": [self.user1.id], "groups": []}, # type: ignore[bad-assignment]
|
||||
},
|
||||
Document,
|
||||
[self.doc1.pk],
|
||||
|
||||
@@ -7,6 +7,7 @@ from documents.data_models import DocumentMetadataOverrides
|
||||
from documents.models import CustomFieldInstance
|
||||
from documents.models import Document
|
||||
from documents.models import WorkflowAction
|
||||
from documents.permissions import PermissionsSpec
|
||||
from documents.permissions import set_permissions_for_objects
|
||||
from documents.templating.workflows import parse_w_workflow_placeholders
|
||||
|
||||
@@ -73,7 +74,7 @@ def apply_assignment_to_document(
|
||||
action.has_assign_change_groups,
|
||||
],
|
||||
):
|
||||
permissions = {
|
||||
permissions: PermissionsSpec = {
|
||||
"view": {
|
||||
"users": action.assign_view_users.values_list("id", flat=True),
|
||||
"groups": action.assign_view_groups.values_list("id", flat=True),
|
||||
@@ -244,7 +245,7 @@ def apply_removal_to_document(
|
||||
document.owner = None
|
||||
|
||||
if action.remove_all_permissions:
|
||||
permissions = {
|
||||
permissions: PermissionsSpec = {
|
||||
"view": {"users": [], "groups": []},
|
||||
"change": {"users": [], "groups": []},
|
||||
}
|
||||
|
||||
Reference in new issue
Block a user