Source code for O365.groups

import logging

from dateutil.parser import parse
from .utils import ApiComponent
from .directory import User

log = logging.getLogger(__name__)


[docs] class Group(ApiComponent): """ A Microsoft O365 group """ _endpoints = { 'get_group_owners': '/groups/{group_id}/owners', 'get_group_members': '/groups/{group_id}/members', } member_constructor = User
[docs] def __init__(self, *, parent=None, con=None, **kwargs): """ A Microsoft O365 group :param parent: parent object :type parent: Teams :param Connection con: connection to use if no parent specified :param Protocol protocol: protocol to use if no parent specified (kwargs) :param str main_resource: use this resource instead of parent resource (kwargs) """ if parent and con: raise ValueError('Need a parent or a connection but not both') self.con = parent.con if parent else con cloud_data = kwargs.get(self._cloud_data_key, {}) self.object_id = cloud_data.get('id') # Choose the main_resource passed in kwargs over parent main_resource main_resource = kwargs.pop('main_resource', None) or ( getattr(parent, 'main_resource', None) if parent else None) main_resource = '{}{}'.format(main_resource, '') super().__init__( protocol=parent.protocol if parent else kwargs.get('protocol'), main_resource=main_resource) self.type = cloud_data.get('@odata.type') self.display_name = cloud_data.get(self._cc('displayName'), '') self.description = cloud_data.get(self._cc('description'), '') self.mail = cloud_data.get(self._cc('mail'), '') self.mail_nickname = cloud_data.get(self._cc('mailNickname'), '') self.visibility = cloud_data.get(self._cc('visibility'), '')
def __str__(self): return self.__repr__() def __repr__(self): return 'Group: {}'.format(self.display_name) def __eq__(self, other): return self.object_id == other.object_id def __hash__(self): return self.object_id.__hash__()
[docs] def get_group_members(self, recursive=False): """ Returns members of given group :param bool recursive: drill down to users if group has other group as a member :rtype: list[User] """ if recursive: recursive_data = self._get_group_members_raw() for member in recursive_data: if member['@odata.type'] == '#microsoft.graph.group': recursive_members = Groups(con=self.con, protocol=self.protocol).get_group_by_id(member['id'])._get_group_members_raw() recursive_data.extend(recursive_members) return [self.member_constructor(parent=self, **{self._cloud_data_key: lst}) for lst in recursive_data] else: return [self.member_constructor(parent=self, **{self._cloud_data_key: lst}) for lst in self._get_group_members_raw()]
def _get_group_members_raw(self): url = self.build_url(self._endpoints.get('get_group_members').format(group_id=self.object_id)) response = self.con.get(url) if not response: return [] data = response.json() return data.get('value', [])
[docs] def get_group_owners(self): """ Returns owners of given group :rtype: list[User] """ url = self.build_url(self._endpoints.get('get_group_owners').format(group_id=self.object_id)) response = self.con.get(url) if not response: return [] data = response.json() return [self.member_constructor(parent=self, **{self._cloud_data_key: lst}) for lst in data.get('value', [])]
[docs] class Groups(ApiComponent): """ A microsoft groups class In order to use the API following permissions are required. Delegated (work or school account) - Group.Read.All, Group.ReadWrite.All """ _endpoints = { 'get_user_groups': '/users/{user_id}/memberOf', 'get_group_by_id': '/groups/{group_id}', 'get_group_by_mail': '/groups/?$search="mail:{group_mail}"&$count=true', 'list_groups': '/groups', } group_constructor = Group
[docs] def __init__(self, *, parent=None, con=None, **kwargs): """ A Teams object :param parent: parent object :type parent: Account :param Connection con: connection to use if no parent specified :param Protocol protocol: protocol to use if no parent specified (kwargs) :param str main_resource: use this resource instead of parent resource (kwargs) """ if parent and con: raise ValueError('Need a parent or a connection but not both') self.con = parent.con if parent else con # Choose the main_resource passed in kwargs over the host_name main_resource = kwargs.pop('main_resource', '') # defaults to blank resource super().__init__( protocol=parent.protocol if parent else kwargs.get('protocol'), main_resource=main_resource)
def __str__(self): return self.__repr__() def __repr__(self): return 'Microsoft O365 Group parent class'
[docs] def get_group_by_id(self, group_id = None): """ Returns Microsoft O365/AD group with given id :param group_id: group id of group :rtype: Group """ if not group_id: raise RuntimeError('Provide the group_id') if group_id: # get channels by the team id url = self.build_url( self._endpoints.get('get_group_by_id').format(group_id=group_id)) response = self.con.get(url) if not response: return None data = response.json() return self.group_constructor(parent=self, **{self._cloud_data_key: data})
[docs] def get_group_by_mail(self, group_mail = None): """ Returns Microsoft O365/AD group by mail field :param group_name: mail of group :rtype: Group """ if not group_mail: raise RuntimeError('Provide the group mail') if group_mail: # get groups by filter mail url = self.build_url( self._endpoints.get('get_group_by_mail').format(group_mail=group_mail)) response = self.con.get(url, headers={'ConsistencyLevel': 'eventual'}) if not response: return None data = response.json() if '@odata.count' in data and data['@odata.count'] < 1: raise RuntimeError('Not found group with provided filters') # mail is unique field so, we expect exact match -> always use first element from list return self.group_constructor(parent=self, **{self._cloud_data_key: data.get('value')[0]})
[docs] def get_user_groups(self, user_id = None): """ Returns list of groups that given user has membership :param user_id: user_id :rtype: list[Group] """ if not user_id: raise RuntimeError('Provide the user_id') if user_id: # get channels by the team id url = self.build_url( self._endpoints.get('get_user_groups').format(user_id=user_id)) response = self.con.get(url) if not response: return None data = response.json() return [ self.group_constructor(parent=self, **{self._cloud_data_key: group}) for group in data.get('value', [])]
[docs] def list_groups(self): """ Returns list of groups :rtype: list[Group] """ url = self.build_url( self._endpoints.get('list_groups')) response = self.con.get(url) if not response: return None data = response.json() return [ self.group_constructor(parent=self, **{self._cloud_data_key: group}) for group in data.get('value', [])]