Source code for globus_sdk.services.groups.client

from typing import Any, Callable, Dict, Optional, TypeVar, Union

from globus_sdk import client, response, utils
from globus_sdk.scopes import GroupsScopes
from globus_sdk.types import UUIDLike

from .data import BatchMembershipActions, GroupPolicies
from .errors import GroupsAPIError

C = TypeVar("C", bound=Callable)


def _groupdoc(message: str, link: str) -> Callable[[C], C]:
    # do not use functools.partial because it doesn't preserve type information
    # see: https://github.com/python/mypy/issues/1484
    def partial(func: C) -> C:
        return utils.doc_api_method(
            message,
            link,
            external_base_url="https://groups.api.globus.org/redoc#operation",
        )(func)

    return partial


[docs]class GroupsClient(client.BaseClient): """ Client for the `Globus Groups API <https://docs.globus.org/api/groups/>`_. This provides a relatively low level client to public groups API endpoints. You may also consider looking at the GroupsManager as a simpler interface to more common actions. .. automethodlist:: globus_sdk.GroupsClient """ base_path = "/v2/" error_class = GroupsAPIError service_name = "groups" scopes = GroupsScopes
[docs] @_groupdoc( "Retrieve your groups and membership", "get_my_groups_and_memberships_v2_groups_my_groups_get", ) def get_my_groups( self, *, query_params: Optional[Dict[str, Any]] = None ) -> response.GlobusHTTPResponse: """ Return a list of groups your identity belongs to. """ return self.get("/groups/my_groups", query_params=query_params)
[docs] @_groupdoc("Get Group", "get_group_v2_groups__group_id__get") def get_group( self, group_id: UUIDLike, *, query_params: Optional[Dict[str, Any]] = None ) -> response.GlobusHTTPResponse: """ Get details about a specific group """ return self.get(f"/groups/{group_id}", query_params=query_params)
[docs] @_groupdoc("Delete a group", "delete_group_v2_groups__group_id__delete") def delete_group( self, group_id: UUIDLike, *, query_params: Optional[Dict[str, Any]] = None ) -> response.GlobusHTTPResponse: """ Delete a group. """ return self.delete(f"/groups/{group_id}", query_params=query_params)
[docs] @_groupdoc("Create a group", "create_group_v2_groups_post") def create_group( self, data: Dict[str, Any], *, query_params: Optional[Dict[str, Any]] = None ) -> response.GlobusHTTPResponse: """ Create a group. """ return self.post("/groups", data=data, query_params=query_params)
[docs] @_groupdoc( "Get the policies for the group", "get_policies_v2_groups__group_id__policies_get", ) def get_group_policies( self, group_id: UUIDLike, *, query_params: Optional[Dict[str, Any]] = None ) -> response.GlobusHTTPResponse: """ Get policies for the given group """ return self.get(f"/groups/{group_id}/policies", query_params=query_params)
[docs] @_groupdoc( "Set the policies for the group", "update_policies_v2_groups__group_id__policies_put", ) def set_group_policies( self, group_id: UUIDLike, data: Union[Dict[str, Any], GroupPolicies], *, query_params: Optional[Dict[str, Any]] = None, ) -> response.GlobusHTTPResponse: """ Set policies for the group. """ return self.put( f"/groups/{group_id}/policies", data=data, query_params=query_params )
[docs] @_groupdoc( "Get the preferences for your identity set", "get_identity_set_preferences_v2_preferences_get", ) def get_identity_preferences( self, *, query_params: Optional[Dict[str, Any]] = None ) -> response.GlobusHTTPResponse: """ Get identity preferences. Currently this only includes whether the user allows themselves to be added to groups. """ return self.get("/preferences", query_params=query_params)
[docs] @_groupdoc( "Set the preferences for your identity set", "put_identity_set_preferences_v2_preferences_put", ) def set_identity_preferences( self, data: Dict[str, Any], *, query_params: Optional[Dict[str, Any]] = None, ) -> response.GlobusHTTPResponse: """ Set identity preferences. Currently this only includes whether the user allows themselves to be added to groups. """ return self.put("/preferences", data=data, query_params=query_params)
[docs] @_groupdoc( "Get the membership fields for your identity set", "get_membership_fields_v2_groups__group_id__membership_fields_get", ) def get_membership_fields( self, group_id: UUIDLike, *, query_params: Optional[Dict[str, Any]] = None, ) -> response.GlobusHTTPResponse: """ Get membership fields for your identities. """ return self.get( f"/groups/{group_id}/membership_fields", query_params=query_params )
[docs] @_groupdoc( "Set the membership fields for your identity set", "put_membership_fields_v2_groups__group_id__membership_fields_put", ) def set_membership_fields( self, group_id: UUIDLike, data: Dict[Any, str], *, query_params: Optional[Dict[str, Any]] = None, ) -> response.GlobusHTTPResponse: """ Get membership fields for your identities. """ return self.put( f"/groups/{group_id}/membership_fields", data=data, query_params=query_params, )
[docs] @_groupdoc( "Perform actions on members of the group", "group_membership_post_actions_v2_groups__group_id__post", ) def batch_membership_action( self, group_id: UUIDLike, actions: Union[Dict[str, Any], BatchMembershipActions], *, query_params: Optional[Dict[str, Any]] = None, ) -> response.GlobusHTTPResponse: """ Execute a batch of actions against several group memberships. """ return self.post(f"/groups/{group_id}", data=actions, query_params=query_params)