- Source code for O365.account
+from .connection import Connection, Protocol, MSGraphProtocol
+from .utils import ME_RESOURCE
+
+
+[docs]class Account:
+
+ connection_constructor = Connection
+
+[docs] def __init__(self, credentials, *, protocol=None, main_resource=None, **kwargs):
+ """ Creates an object which is used to access resources related to the
+ specified credentials
+
+ :param tuple credentials: a tuple containing the client_id
+ and client_secret
+ :param Protocol protocol: the protocol to be used in this account
+ :param str main_resource: the resource to be used by this account
+ ('me' or 'users', etc.)
+ :param kwargs: any extra args to be passed to the Connection instance
+ :raises ValueError: if an invalid protocol is passed
+ """
+
+ protocol = protocol or MSGraphProtocol # Defaults to Graph protocol
+ self.protocol = protocol(default_resource=main_resource,
+ **kwargs) if isinstance(protocol,
+ type) else protocol
+
+ if not isinstance(self.protocol, Protocol):
+ raise ValueError("'protocol' must be a subclass of Protocol")
+
+ auth_flow_type = kwargs.get('auth_flow_type', 'authorization')
+ scopes = kwargs.get('scopes', None) # retrieve scopes
+
+ if auth_flow_type in ('authorization', 'public'):
+ # convert the provided scopes to protocol scopes:
+ if scopes is not None:
+ kwargs['scopes'] = self.protocol.get_scopes_for(scopes)
+ elif auth_flow_type == 'credentials':
+ # for client credential grant flow solely:
+ # append the default scope if it's not provided
+ if not scopes:
+ kwargs['scopes'] = [self.protocol.prefix_scope('.default')]
+
+ # set main_resource to blank when it's the 'ME' resource
+ if self.protocol.default_resource == ME_RESOURCE:
+ self.protocol.default_resource = ''
+ if main_resource == ME_RESOURCE:
+ main_resource = ''
+ else:
+ raise ValueError('"auth_flow_type" must be "authorization", "credentials" or "public"')
+
+ self.con = self.connection_constructor(credentials, **kwargs)
+ self.main_resource = main_resource or self.protocol.default_resource
+
+ def __repr__(self):
+ if self.con.auth:
+ return 'Account Client Id: {}'.format(self.con.auth[0])
+ else:
+ return 'Unidentified Account'
+
+ @property
+ def is_authenticated(self):
+ """
+ Checks whether the library has the authentication and that is not expired
+ :return: True if authenticated, False otherwise
+ """
+ token = self.con.token_backend.token
+ if not token:
+ token = self.con.token_backend.get_token()
+
+ return token is not None and not token.is_expired
+
+[docs] def authenticate(self, *, scopes=None, **kwargs):
+ """ Performs the oauth authentication flow using the console resulting in a stored token.
+ It uses the credentials passed on instantiation
+
+ :param list[str] or None scopes: list of protocol user scopes to be converted
+ by the protocol or scope helpers
+ :param kwargs: other configurations to be passed to the
+ Connection.get_authorization_url and Connection.request_token methods
+ :return: Success / Failure
+ :rtype: bool
+ """
+
+ if self.con.auth_flow_type in ('authorization', 'public'):
+ if scopes is not None:
+ if self.con.scopes is not None:
+ raise RuntimeError('The scopes must be set either at the Account instantiation or on the account.authenticate method.')
+ self.con.scopes = self.protocol.get_scopes_for(scopes)
+ else:
+ if self.con.scopes is None:
+ raise ValueError('The scopes are not set. Define the scopes requested.')
+
+ consent_url, _ = self.con.get_authorization_url(**kwargs)
+
+ print('Visit the following url to give consent:')
+ print(consent_url)
+
+ token_url = input('Paste the authenticated url here:\n')
+
+ if token_url:
+ result = self.con.request_token(token_url, **kwargs) # no need to pass state as the session is the same
+ if result:
+ print('Authentication Flow Completed. Oauth Access Token Stored. You can now use the API.')
+ else:
+ print('Something go wrong. Please try again.')
+
+ return bool(result)
+ else:
+ print('Authentication Flow aborted.')
+ return False
+
+ elif self.con.auth_flow_type == 'credentials':
+ return self.con.request_token(None, requested_scopes=scopes)
+ else:
+ raise ValueError('Connection "auth_flow_type" must be "authorization", "public" or "credentials"')
+
+[docs] def get_current_user(self):
+ """ Returns the current user """
+ if self.con.auth_flow_type in ('authorization', 'public'):
+ directory = self.directory(resource=ME_RESOURCE)
+ return directory.get_current_user()
+ else:
+ return None
+
+ @property
+ def connection(self):
+ """ Alias for self.con
+
+ :rtype: type(self.connection_constructor)
+ """
+ return self.con
+
+[docs] def new_message(self, resource=None):
+ """ Creates a new message to be sent or stored
+
+ :param str resource: Custom resource to be used in this message
+ (Defaults to parent main_resource)
+ :return: New empty message
+ :rtype: Message
+ """
+ from .message import Message
+ return Message(parent=self, main_resource=resource, is_draft=True)
+
+[docs] def mailbox(self, resource=None):
+ """ Get an instance to the mailbox for the specified account resource
+
+ :param str resource: Custom resource to be used in this mailbox
+ (Defaults to parent main_resource)
+ :return: a representation of account mailbox
+ :rtype: O365.mailbox.MailBox
+ """
+ from .mailbox import MailBox
+ return MailBox(parent=self, main_resource=resource, name='MailBox')
+
+[docs] def address_book(self, *, resource=None, address_book='personal'):
+ """ Get an instance to the specified address book for the
+ specified account resource
+
+ :param str resource: Custom resource to be used in this address book
+ (Defaults to parent main_resource)
+ :param str address_book: Choose from 'Personal' or 'Directory'
+ :return: a representation of the specified address book
+ :rtype: AddressBook or GlobalAddressList
+ :raises RuntimeError: if invalid address_book is specified
+ """
+ if address_book.lower() == 'personal':
+ from .address_book import AddressBook
+
+ return AddressBook(parent=self, main_resource=resource,
+ name='Personal Address Book')
+ elif address_book.lower() in ('gal', 'directory'):
+ # for backwards compatibility only
+ from .directory import Directory
+
+ return Directory(parent=self, main_resource=resource)
+ else:
+ raise RuntimeError(
+ 'address_book must be either "Personal" '
+ '(resource address book) or "Directory" (Active Directory)')
+
+[docs] def directory(self, resource=None):
+ """ Returns the active directory instance"""
+ from .directory import Directory, USERS_RESOURCE
+
+ return Directory(parent=self, main_resource=resource or USERS_RESOURCE)
+
+[docs] def schedule(self, *, resource=None):
+ """ Get an instance to work with calendar events for the
+ specified account resource
+
+ :param str resource: Custom resource to be used in this schedule object
+ (Defaults to parent main_resource)
+ :return: a representation of calendar events
+ :rtype: Schedule
+ """
+ from .calendar import Schedule
+ return Schedule(parent=self, main_resource=resource)
+
+[docs] def storage(self, *, resource=None):
+ """ Get an instance to handle file storage (OneDrive / Sharepoint)
+ for the specified account resource
+
+ :param str resource: Custom resource to be used in this drive object
+ (Defaults to parent main_resource)
+ :return: a representation of OneDrive File Storage
+ :rtype: Storage
+ :raises RuntimeError: if protocol doesn't support the feature
+ """
+ if not isinstance(self.protocol, MSGraphProtocol):
+ # TODO: Custom protocol accessing OneDrive/Sharepoint Api fails here
+ raise RuntimeError(
+ 'Drive options only works on Microsoft Graph API')
+ from .drive import Storage
+ return Storage(parent=self, main_resource=resource)
+
+
+
+[docs] def planner(self, *, resource=''):
+ """ Get an instance to read information from Microsoft planner """
+
+ if not isinstance(self.protocol, MSGraphProtocol):
+ # TODO: Custom protocol accessing OneDrive/Sharepoint Api fails here
+ raise RuntimeError(
+ 'planner api only works on Microsoft Graph API')
+
+ from .planner import Planner
+ return Planner(parent=self, main_resource=resource)
+
+[docs] def teams(self, *, resource=''):
+ """ Get an instance to read information from Microsoft Teams """
+
+ if not isinstance(self.protocol, MSGraphProtocol):
+ raise RuntimeError(
+ 'teams api only works on Microsoft Graph API')
+
+ from .teams import Teams
+ return Teams(parent=self, main_resource=resource)
+
+[docs] def outlook_categories(self, *, resource=''):
+ """ Returns a Categories object to handle the available Outlook Categories """
+ from .category import Categories
+
+ return Categories(parent=self, main_resource=resource)
+