search_param: refactor SharedOrMyDraftsParam

invenio_rdm_records/services/search_params.py

@@ -13,7 +13,7 @@
 from invenio_access.permissions import authenticated_user
 from invenio_records_resources.services.records.params.base import ParamInterpreter
 
-from invenio_rdm_records.records.systemfields.access.grants import Grant
+from invenio_rdm_records.services.generators import AccessGrant
 from invenio_rdm_records.records.systemfields.deletion_status import (
     RecordDeletionStatusEnum,
 )
@@ -51,32 +51,12 @@ class SharedOrMyDraftsParam(ParamInterpreter):
     Returns only drafts owned by the user or shared with the user via grant subject user or role.
     """
 
-    def _make_grant_token(self, subj_type, subj_id, permission):
-        """Create a grant token from the specified parts."""
-        # NOTE: `Grant.to_token()` doesn't need the actual subject to be set
-        return Grant(
-            subject=None,
-            origin=None,
-            permission=permission,
-            subject_type=subj_type,
-            subject_id=subj_id,
-        ).to_token()
-
-    def _grant_tokens(self, identity, permissions):
-        """Parse a list of grant tokens provided by the given identity."""
+    def _generate_shared_with_me_query(self, identity):
+        """Generate the shared_with_me query."""
         tokens = []
-        for _permission in permissions:
-            for need in identity.provides:
-                token = None
-                if need.method == "id":
-                    token = self._make_grant_token("user", need.value, _permission)
-                elif need.method == "role":
-                    token = self._make_grant_token("role", need.value, _permission)
-
-                if token is not None:
-                    tokens.append(token)
-
-        return tokens
+        for _permission in ["preview", "edit", "manage"]:
+            tokens.extend(AccessGrant(_permission)._grant_tokens(identity))
+        return dsl.Q("terms", **{"parent.access.grant_tokens": tokens})
 
     def apply(self, identity, search, params):
         """Evaluate the include_deleted parameter on the search."""
@@ -90,13 +70,8 @@ def is_user_authenticated():
         if value is None and is_user_authenticated():
             if params.get("shared_with_me") is True:
                 # Shared with me
-                tokens = self._grant_tokens(
-                    identity, permissions=["preview", "edit", "manage"]
-                )
-                shared_with_me = dsl.Q(
-                    "terms", **{"parent.access.grant_tokens": tokens}
-                )
-                return search.filter(shared_with_me)
+                shared_with_me = self._generate_shared_with_me_query(identity)
+                search = search.filter(shared_with_me)
             elif params.get("shared_with_me") is False:
                 # My uploads
                 my_uploads = dsl.Q(