Connection

exception O365.connection.TokenExpiredError(*args, **kwargs)[source]: Bases: HTTPError

class O365.connection.Connection(credentials: tuple, *, proxy_server: str | None = None, proxy_port: int | None = 8080, proxy_username: str | None = None, proxy_password: str | None = None, proxy_http_only: bool = False, requests_delay: int = 200, raise_http_errors: bool = True, request_retries: int = 3, token_backend: BaseTokenBackend | None = None, tenant_id: str = 'common', auth_flow_type: str = 'authorization', username: str | None = None, password: str | None = None, timeout: int | None = None, json_encoder: JSONEncoder | None = None, verify_ssl: bool = True, default_headers: dict = None, store_token_after_refresh: bool = True, **kwargs)[source]

Bases: object

Handles all communication (requests) between the app and the server

__init__(credentials: tuple, *, proxy_server: str | None = None, proxy_port: int | None = 8080, proxy_username: str | None = None, proxy_password: str | None = None, proxy_http_only: bool = False, requests_delay: int = 200, raise_http_errors: bool = True, request_retries: int = 3, token_backend: BaseTokenBackend | None = None, tenant_id: str = 'common', auth_flow_type: str = 'authorization', username: str | None = None, password: str | None = None, timeout: int | None = None, json_encoder: JSONEncoder | None = None, verify_ssl: bool = True, default_headers: dict = None, store_token_after_refresh: bool = True, **kwargs)[source]

Creates an API connection object

Parameters:

credentials (tuple) – a tuple of (client_id, client_secret) Generate client_id and client_secret in https://entra.microsoft.com/
proxy_server (str) – the proxy server
proxy_port (int) – the proxy port, defaults to 8080
proxy_username (str) – the proxy username
proxy_password (str) – the proxy password
requests_delay (int) – number of milliseconds to wait between api calls. The Api will respond with 429 Too many requests if more than 17 requests are made per second. Defaults to 200 milliseconds just in case more than 1 connection is making requests across multiple processes.
raise_http_errors (bool) – If True Http 4xx and 5xx status codes will raise as exceptions
request_retries (int) – number of retries done when the server responds with 5xx error codes.
token_backend (BaseTokenBackend) – the token backend used to get and store tokens
tenant_id (str) – use this specific tenant id, defaults to common
default_headers (dict) – allow to force headers in api call (ex: default_headers={“Prefer”: ‘IdType=”ImmutableId”’}) to get constant id for objects.
auth_flow_type (str) –
the auth method flow style used: Options:
- ’authorization’: 2-step web style grant flow using an authentication url
- ’public’: 2-step web style grant flow using an authentication url for public apps where
  client secret cannot be secured
- ’credentials’: also called client credentials grant flow using only the client id and secret.
  The secret can be certificate based authentication
- ’password’: using the username and password. Not recommended
username (str) – The username the credentials will be taken from in the token backend. If None, the username will be the first one found in the token backend. The user’s email address to provide in case of auth_flow_type == ‘password’
password (str) – The user’s password to provide in case of auth_flow_type == ‘password’
timeout (float or tuple) – How long to wait for the server to send data before giving up, as a float, or a tuple (connect timeout, read timeout)
json_encoder (JSONEncoder) – The JSONEncoder to use during the JSON serialization on the request.
verify_ssl (bool) – set the verify flag on the requests library
store_token_after_refresh (bool) – if after a token refresh the token backend should call save_token
kwargs (dict) – any extra params passed to Connection

Raises:

ValueError – if credentials is not tuple of (client_id, client_secret)

delete(url: str, **kwargs) → Response[source]

Shorthand for self.request(url, ‘delete’)

Parameters:

url (str) – url to send delete oauth request to
kwargs – extra params to send to request api

Returns:

Response of the request

Return type:

requests.Response

get(url: str, params: dict | None = None, **kwargs) → Response[source]

Shorthand for self.oauth_request(url, ‘get’)

Parameters:

url (str) – url to send get oauth request to
params (dict) – request parameter to get the service data
kwargs – extra params to send to request api

Returns:

Response of the request

Return type:

requests.Response

get_authorization_url(requested_scopes: List[str], redirect_uri: str | None = None, **kwargs) → tuple[str, dict][source]

Initializes the oauth authorization flow, getting the authorization url that the user must approve.

Parameters:

requested_scopes (list[str]) – list of scopes to request access for
redirect_uri (str) – redirect url configured in registered app
kwargs – allow to pass unused params in conjunction with Connection

Returns:

authorization url and the flow dict

get_naive_session() → Session[source]: Creates and returns a naive session

get_session(load_token: bool = False) → Session[source]

Create a requests Session object with the oauth token attached to it

Parameters:: load_token (bool) – load the token from the token backend and load the access token into the session auth
Returns:: A ready to use requests session with authentication header attached
Return type:: requests.Session

load_token_from_backend() → bool[source]: Loads the token from the backend and tries to set the self.username if it’s not set

naive_request(url: str, method: str, **kwargs) → Response[source]

Makes a request to url using an without oauth authorization session, but through a normal session

Parameters:

url (str) – url to send request to
method (str) – type of request (get/put/post/patch/delete)
kwargs – extra params to send to the request api

Returns:

Response of the request

Return type:

requests.Response

oauth_request(url: str, method: str, **kwargs) → Response[source]

Makes a request to url using an oauth session. Raises RuntimeError if the session does not have an Authorization header

Parameters:

url (str) – url to send request to
method (str) – type of request (get/put/post/patch/delete)
kwargs – extra params to send to the request api

Returns:

Response of the request

Return type:

requests.Response

patch(url: str, data: dict | None = None, **kwargs) → Response[source]

Shorthand for self.oauth_request(url, ‘patch’)

Parameters:

url (str) – url to send patch oauth request to
data (dict) – patch data to update the service
kwargs – extra params to send to request api

Returns:

Response of the request

Return type:

requests.Response

post(url: str, data: dict | None = None, **kwargs) → Response[source]

Shorthand for self.oauth_request(url, ‘post’)

Parameters:

url (str) – url to send post oauth request to
data (dict) – post data to update the service
kwargs – extra params to send to request api

Returns:

Response of the request

Return type:

requests.Response

put(url: str, data: dict | None = None, **kwargs) → Response[source]

Shorthand for self.oauth_request(url, ‘put’)

Parameters:

url (str) – url to send put oauth request to
data (dict) – put data to update the service
kwargs – extra params to send to request api

Returns:

Response of the request

Return type:

requests.Response

refresh_token() → bool[source]

Refresh the OAuth authorization token. This will be called automatically when the access token expires, however, you can manually call this method to request a new refresh token.

Return bool:: Success / Failure

request_token(authorization_url: str | None, *, flow: dict | None = None, requested_scopes: List[str] | None = None, store_token: bool = True, **kwargs) → bool[source]

Authenticates for the specified url and gets the oauth token data. Saves the token in the backend if store_token is True. This will replace any other tokens stored for the same username and scopes requested. If the token data is successfully requested, then this method will try to set the username if not previously set.

Parameters:

authorization_url (str or None) – url given by the authorization flow or None if it’s client credentials
flow (dict) – dict object holding the data used in get_authorization_url
requested_scopes (list[str]) – list of scopes to request access for
store_token (bool) – True to store the token in the token backend, so you don’t have to keep opening the auth link and authenticating every time
kwargs – allow to pass unused params in conjunction with Connection

Returns:

Success/Failure

Return type:

bool

set_proxy(proxy_server: str, proxy_port: int, proxy_username: str, proxy_password: str, proxy_http_only: bool) → None[source]

Sets a proxy on the Session

Parameters:

proxy_server (str) – the proxy server
proxy_port (int) – the proxy port, defaults to 8080
proxy_username (str) – the proxy username
proxy_password (str) – the proxy password
proxy_http_only (bool) – if the proxy should only be used for http

update_session_auth_header(access_token: str | None = None) → None[source]: Will update the internal request session auth header with an access token

auth: tuple

The credentials for the connection.

Type: tuple

property auth_flow_type: str

default_headers: Dict

The default headers.

Type: dict

json_encoder: JSONEncoder | None

JSONEncoder to use.

Type: json.JSONEncoder

property msal_client: PublicClientApplication | ConfidentialClientApplication: Returns the msal client or creates it if it’s not already done

naive_session: Session | None

the naive session.

Type: Session

oauth_redirect_url: str

The oauth redirect url.

Type: str

password: str | None

The password for the connection.

Type: str

proxy: Dict

The proxy to use.

Type: dict

raise_http_errors: bool

Should http errors be raised. Default true.

Type: bool

request_retries: int

Number of time to retry request. Default 3.

Type: int

requests_delay: int

The delay to put in a request. Default 0.

Type: int

session: Session | None

The session to use.

Type: Session

store_token_after_refresh: bool

Store token after refresh. Default true.

Type: bool

tenant_id: str

The tenant id.

Type: str

timeout: int

Timeout for the request. Default None.

Type: int

token_backend: BaseTokenBackend

The token backend in use.

Type: BaseTokenbackend

property username: str | None: Returns the username in use If username is not set this will try to set the username to the first account found from the token_backend.

verify_ssl: bool

Whether to verify the ssl cert. Default true.

Type: bool

class O365.connection.MSBusinessCentral365Protocol(api_version: str = 'v1.0', default_resource: str | None = None, environment: str | None = None, **kwargs)[source]

Bases: Protocol

A Microsoft Business Central Protocol Implementation https://docs.microsoft.com/en-us/dynamics-nav/api-reference/v1.0/endpoints-apis-for-dynamics

__init__(api_version: str = 'v1.0', default_resource: str | None = None, environment: str | None = None, **kwargs)[source]

Create a new Microsoft Graph protocol object

_protocol_url = ‘https://api.businesscentral.dynamics.com/’

_oauth_scope_prefix = ‘https://api.businesscentral.dynamics.com/’

Parameters:

api_version (str) – api version to use
default_resource (str) – the default resource to use when there is nothing explicitly specified during the requests

max_top_value: int

The max value for ‘top’ (999).

Type: str

property timezone: ZoneInfo

class O365.connection.MSGraphProtocol(api_version: str = 'v1.0', default_resource: str | None = None, **kwargs)[source]

Bases: Protocol

A Microsoft Graph Protocol Implementation https://docs.microsoft.com/en-us/outlook/rest/compare-graph-outlook

__init__(api_version: str = 'v1.0', default_resource: str | None = None, **kwargs)[source]

Create a new Microsoft Graph protocol object

_protocol_url = ‘https://graph.microsoft.com/’

_oauth_scope_prefix = ‘https://graph.microsoft.com/’

Parameters:

api_version (str) – api version to use
default_resource (str) – the default resource to use when there is nothing explicitly specified during the requests

max_top_value: int

The max value for ‘top’ (999).

Type: str

property timezone: ZoneInfo

class O365.connection.MSOffice365Protocol(api_version: str = 'v2.0', default_resource: str | None = None, **kwargs)[source]

Bases: Protocol

A Microsoft Office 365 Protocol Implementation https://docs.microsoft.com/en-us/outlook/rest/compare-graph-outlook

__init__(api_version: str = 'v2.0', default_resource: str | None = None, **kwargs)[source]

Create a new Office 365 protocol object

_protocol_url = ‘https://outlook.office.com/api/’

_oauth_scope_prefix = ‘https://outlook.office.com/’

Parameters:

api_version (str) – api version to use
default_resource (str) – the default resource to use when there is nothing explicitly specified during the requests

max_top_value: int

The max value for ‘top’ (999).

Type: str

property timezone: ZoneInfo

Bases: object

Base class for all protocols

static to_api_case(key: str) → str[source]

Converts key to snake_case

Parameters:: key – key to convert into snake_case
Returns:: key after case conversion

Create a new protocol object

Parameters:

protocol_url – the base url used to communicate with the server
api_version – the api version
default_resource – the default resource to use when there is nothing explicitly specified during the requests
casing_function – the casing transform function to be used on api keywords (camelcase / pascalcase)
protocol_scope_prefix – prefix url for scopes
timezone – preferred timezone, if not provided will default to the system timezone or fallback to UTC

Raises:

ValueError – if protocol_url or api_version are not supplied

convert_case(key: str) → str[source]

Returns a key converted with this protocol casing method

Converts case to send/read from the cloud

When using Microsoft Graph API, the keywords of the API use lowerCamelCase Casing

When using Office 365 API, the keywords of the API use PascalCase Casing

Default case in this API is lowerCamelCase

Parameters:: key – a dictionary key to convert
Returns:: key after case conversion

get_scopes_for(user_provided_scopes: list | str | tuple | None) → list[source]

Returns a list of scopes needed for each of the scope_helpers provided, by adding the prefix to them if required

Parameters:: user_provided_scopes – a list of scopes or scope helpers
Returns:: scopes with url prefix added
Raises:: ValueError – if unexpected datatype of scopes are passed

get_service_keyword(keyword: str) → str | None[source]

Returns the data set to the key in the internal data-key dict

Parameters:: keyword – key to get value for
Returns:: value of the keyword

prefix_scope(scope: str) → str[source]: Inserts the protocol scope prefix if required

api_version: str

The api version being used.

Type: str

casing_function: Callable

The casing function being used.

Type: callable

default_resource: str

The resource being used. Defaults to ‘me’.

Type: str

keyword_data_store: dict

The keyword data store.

Type: dict

max_top_value: int

The max value for ‘top’ (500).

Type: str

protocol_scope_prefix: str

The scope prefix for protcol in use.

Type: str

protocol_url: str

The url for the protcol in use.

Type: str

service_url: str

The full service url.

Type: str

property timezone: ZoneInfo

use_default_casing: bool

Indicates if default casing is being used.

Type: bool

O365.connection.oauth_authentication_flow(client_id: str, client_secret: str, scopes: List[str] = None, protocol: Protocol | None = None, **kwargs) → bool[source]

A helper method to perform the OAuth2 authentication flow. Authenticate and get the oauth token

Parameters:

client_id (str) – the client_id
client_secret (str) – the client_secret
scopes (list[str]) – a list of protocol user scopes to be converted by the protocol or raw scopes
protocol (Protocol) – the protocol to be used. Defaults to MSGraphProtocol
kwargs – other configuration to be passed to the Connection instance, connection.get_authorization_url or connection.request_token

Returns:

Success or Failure

Return type:

bool