Creation of a dry_run flag in delete_permissions and listing the ACL tree given a container or entity

Author: BryanFauble

synapseclient/api/__init__.py

@@ -29,6 +29,7 @@
     delete_entity_generated_by,
     get_entities_by_md5,
     get_entity,
+    get_entity_acl,
     get_entity_path,
     get_upload_destination,
     get_upload_destination_location,
@@ -55,6 +56,8 @@
     get_default_columns,
     post_columns,
 )
+from .team_services import post_team_list
+from .user_services import get_user_group_headers_batch
 
 __all__ = [
     # annotations
@@ -83,6 +86,7 @@
     "post_entity",
     "delete_entity",
     "delete_entity_acl",
+    "get_entity_acl",
     "get_upload_destination",
     "get_upload_destination_location",
     "create_access_requirements_if_none",
@@ -110,4 +114,8 @@
     "get_default_columns",
     "ViewTypeMask",
     "ViewEntityType",
+    # team_services
+    "post_team_list",
+    # user_services
+    "get_user_group_headers_batch",
 ]

synapseclient/api/entity_services.py

@@ -3,6 +3,7 @@
 """
 
 import json
+from dataclasses import dataclass
 from typing import TYPE_CHECKING, Any, Dict, List, Optional, Union
 
 from async_lru import alru_cache
@@ -13,6 +14,74 @@
     from synapseclient import Synapse
 
 
+@dataclass
+class EntityHeader:
+    """
+    JSON schema for EntityHeader POJO. This represents metadata about a Synapse entity.
+
+    Attributes:
+        name: The name of the entity
+        id: The id of the entity
+        type: The type of the entity
+        version_number: The version number of the entity
+        version_label: The user defined version label of the entity
+        is_latest_version: If this version is the latest version of the entity
+        benefactor_id: The ID of the entity that this Entity's ACL is inherited from
+        created_on: The date this entity was created
+        modified_on: The date this entity was last modified
+        created_by: The ID of the user that created this entity
+        modified_by: The ID of the user that last modified this entity
+    """
+
+    name: Optional[str] = None
+    """The name of the entity"""
+
+    id: Optional[str] = None
+    """The id of the entity"""
+
+    type: Optional[str] = None
+    """The type of the entity"""
+
+    version_number: Optional[int] = None
+    """The version number of the entity"""
+
+    version_label: Optional[str] = None
+    """The user defined version label of the entity"""
+
+    is_latest_version: Optional[bool] = None
+    """If this version is the latest version of the entity"""
+
+    benefactor_id: Optional[int] = None
+    """The ID of the entity that this Entity's ACL is inherited from"""
+
+    created_on: Optional[str] = None
+    """The date this entity was created"""
+
+    modified_on: Optional[str] = None
+    """The date this entity was last modified"""
+
+    created_by: Optional[str] = None
+    """The ID of the user that created this entity"""
+
+    modified_by: Optional[str] = None
+    """The ID of the user that last modified this entity"""
+
+    def fill_from_dict(self, synapse_response: Dict[str, Any]) -> "EntityHeader":
+        """Converts a response from the REST API into this dataclass."""
+        self.name = synapse_response.get("name", None)
+        self.id = synapse_response.get("id", None)
+        self.type = synapse_response.get("type", None)
+        self.version_number = synapse_response.get("versionNumber", None)
+        self.version_label = synapse_response.get("versionLabel", None)
+        self.is_latest_version = synapse_response.get("isLatestVersion", None)
+        self.benefactor_id = synapse_response.get("benefactorId", None)
+        self.created_on = synapse_response.get("createdOn", None)
+        self.modified_on = synapse_response.get("modifiedOn", None)
+        self.created_by = synapse_response.get("createdBy", None)
+        self.modified_by = synapse_response.get("modifiedBy", None)
+        return self
+
+
 async def post_entity(
     request: Dict[str, Any],
     generated_by: Optional[str] = None,
@@ -345,6 +414,81 @@ async def main():
     )
 
 
+async def get_entity_acl(
+    entity_id: str,
+    *,
+    synapse_client: Optional["Synapse"] = None,
+):
+    """
+    Get the Access Control List (ACL) for an entity.
+
+    Note: If this method is called on an Entity that is inheriting its permission
+    from another Entity a NOT_FOUND (404) response will be generated. The error
+    response message will include the Entity's benefactor ID.
+
+    Arguments:
+        entity_id: The ID of the entity.
+        synapse_client: If not passed in and caching was not disabled by
+                `Synapse.allow_client_caching(False)` this will use the last created
+                instance from the Synapse class constructor.
+    """
+    from synapseclient import Synapse
+
+    client = Synapse.get_client(synapse_client=synapse_client)
+    return await client.rest_get_async(
+        uri=f"/entity/{entity_id}/acl",
+    )
+
+
+async def get_entity_benefactor(
+    entity_id: str,
+    *,
+    synapse_client: Optional["Synapse"] = None,
+) -> EntityHeader:
+    """
+    Get an Entity's benefactor. An Entity gets its ACL from its benefactor.
+
+    Implements:
+    <https://rest-docs.synapse.org/rest/GET/entity/id/benefactor.html>
+
+    Arguments:
+        entity_id: The ID of the entity.
+        synapse_client: If not passed in and caching was not disabled by
+                `Synapse.allow_client_caching(False)` this will use the last created
+                instance from the Synapse class constructor.
+
+    Returns:
+        The EntityHeader of the entity's benefactor (the entity from which it inherits its ACL).
+
+    Example: Get the benefactor of an entity
+        Get the benefactor entity header for entity `syn123`.
+
+        ```python
+        import asyncio
+        from synapseclient import Synapse
+        from synapseclient.api import get_entity_benefactor
+
+        syn = Synapse()
+        syn.login()
+
+        async def main():
+            benefactor = await get_entity_benefactor(entity_id="syn123")
+            print(f"Entity benefactor: {benefactor.name} (ID: {benefactor.id})")
+
+        asyncio.run(main())
+        ```
+    """
+    from synapseclient import Synapse
+
+    client = Synapse.get_client(synapse_client=synapse_client)
+    response = await client.rest_get_async(
+        uri=f"/entity/{entity_id}/benefactor",
+    )
+
+    entity_header = EntityHeader()
+    return entity_header.fill_from_dict(response)
+
+
 async def get_entity_path(
     entity_id: str,
     *,

synapseclient/api/team_services.py

@@ -0,0 +1,50 @@
+"""This module is responsible for exposing the services defined at:
+<https://rest-docs.synapse.org/rest/#org.sagebionetworks.repo.web.controller.TeamController>
+"""
+
+import json
+from typing import TYPE_CHECKING, Dict, List, Optional, Union
+
+if TYPE_CHECKING:
+    from synapseclient import Synapse
+
+
+async def post_team_list(
+    team_ids: List[int],
+    *,
+    synapse_client: Optional["Synapse"] = None,
+) -> Optional[List[Dict[str, Union[str, bool]]]]:
+    """
+    Retrieve a list of Teams given their IDs. Invalid IDs in the list are ignored:
+    The results list is simply smaller than the list of IDs passed in.
+
+    Arguments:
+        team_ids: List of team IDs to retrieve
+        synapse_client: If not passed in and caching was not disabled by
+                `Synapse.allow_client_caching(False)` this will use the last created
+                instance from the Synapse class constructor.
+
+    Returns:
+        List of dictionaries representing <https://rest-docs.synapse.org/rest/org/sagebionetworks/repo/model/Team.html>
+
+    Example: Get teams by their IDs
+        Retrieve specific teams by providing their IDs.
+
+        ```python
+
+        ```
+    """
+    from synapseclient import Synapse
+
+    client = Synapse.get_client(synapse_client=synapse_client)
+
+    request_body = {"list": team_ids}
+
+    response = await client.rest_post_async(
+        uri="/teamList", body=json.dumps(request_body)
+    )
+
+    if "list" in response:
+        return response["list"] or None
+
+    return None

synapseclient/api/user_services.py

@@ -0,0 +1,49 @@
+"""This module is responsible for exposing the services defined at:
+<https://rest-docs.synapse.org/rest/#org.sagebionetworks.repo.web.controller.UserGroupController>
+"""
+
+from typing import TYPE_CHECKING, Dict, List, Optional, Union
+
+if TYPE_CHECKING:
+    from synapseclient import Synapse
+
+
+async def get_user_group_headers_batch(
+    ids: List[str],
+    *,
+    synapse_client: Optional["Synapse"] = None,
+) -> List[Dict[str, Union[str, bool]]]:
+    """
+    Batch get UserGroupHeaders. This fetches information about a collection of
+    users or groups, specified by Synapse IDs.
+
+    Arguments:
+        ids: List of user/group IDs to retrieve
+        synapse_client: If not passed in and caching was not disabled by
+                `Synapse.allow_client_caching(False)` this will use the last created
+                instance from the Synapse class constructor.
+
+    Returns:
+        List representing "children" in
+        <https://rest-docs.synapse.org/rest/org/sagebionetworks/repo/model/UserGroupHeaderResponsePage.html>
+
+    Example: Get user group headers by their IDs
+        Retrieve specific user/group headers by providing their IDs.
+
+        ```python
+
+        ```
+    """
+    from synapseclient import Synapse
+
+    client = Synapse.get_client(synapse_client=synapse_client)
+
+    ids_param = ",".join(ids)
+    params = {"ids": ids_param}
+
+    response = await client.rest_get_async(uri="/userGroupHeaders/batch", params=params)
+
+    if "children" in response:
+        return response["children"] or []
+
+    return []