-
Notifications
You must be signed in to change notification settings - Fork 39
Expand file tree
/
Copy pathclient.py
More file actions
316 lines (262 loc) · 12.8 KB
/
client.py
File metadata and controls
316 lines (262 loc) · 12.8 KB
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
289
290
291
292
293
294
295
296
297
298
299
300
301
302
303
304
305
306
307
308
309
310
311
312
313
314
315
316
from typing import Optional
from urllib.parse import urlencode
import requests
from microsoftgraph import exceptions
from microsoftgraph.calendar import Calendar
from microsoftgraph.contacts import Contacts
from microsoftgraph.files import Files
from microsoftgraph.mail import Mail
from microsoftgraph.notes import Notes
from microsoftgraph.response import Response
from microsoftgraph.users import Users
from microsoftgraph.webhooks import Webhooks
from microsoftgraph.workbooks import Workbooks
class Client(object):
AUTHORITY_URL = "https://login.microsoftonline.com/"
AUTH_ENDPOINT = "/oauth2/v2.0/authorize?"
TOKEN_ENDPOINT = "/oauth2/v2.0/token"
RESOURCE = "https://graph.microsoft.com/"
def __init__(
self,
client_id: str,
client_secret: str,
api_version: str = "v1.0",
account_type: str = "common",
requests_hooks: dict = None,
paginate: bool = True,
) -> None:
"""Instantiates library.
Args:
client_id (str): Application client id.
client_secret (str): Application client secret.
api_version (str, optional): v1.0 or beta. Defaults to "v1.0".
account_type (str, optional): common, organizations or consumers. Defaults to "common".
requests_hooks (dict, optional): Requests library event hooks. Defaults to None.
Raises:
Exception: requests_hooks is not a dict.
"""
self.client_id = client_id
self.client_secret = client_secret
self.api_version = api_version
self.account_type = account_type
self.base_url = self.RESOURCE + self.api_version + "/"
self.token = None
self.workbook_session_id = None
self.paginate = paginate
self.calendar = Calendar(self)
self.contacts = Contacts(self)
self.files = Files(self)
self.mail = Mail(self)
self.notes = Notes(self)
self.users = Users(self)
self.webhooks = Webhooks(self)
self.workbooks = Workbooks(self)
if requests_hooks and not isinstance(requests_hooks, dict):
raise Exception(
'requests_hooks must be a dict. e.g. {"response": func}. http://docs.python-requests.org/en/master/user/advanced/#event-hooks'
)
self.requests_hooks = requests_hooks
def authorization_url(self, redirect_uri: str, scope: list, state: str = None) -> str:
"""Generates an Authorization URL.
The first step to getting an access token for many OpenID Connect (OIDC) and OAuth 2.0 flows is to redirect the
user to the Microsoft identity platform /authorize endpoint. Azure AD will sign the user in and ensure their
consent for the permissions your app requests. In the authorization code grant flow, after consent is obtained,
Azure AD will return an authorization_code to your app that it can redeem at the Microsoft identity platform
/token endpoint for an access token.
https://docs.microsoft.com/en-us/graph/auth-v2-user#2-get-authorization
Args:
redirect_uri (str): The redirect_uri of your app, where authentication responses can be sent and received by
your app. It must exactly match one of the redirect_uris you registered in the app registration portal.
scope (list): A list of the Microsoft Graph permissions that you want the user to consent to. This may also
include OpenID scopes.
state (str, optional): A value included in the request that will also be returned in the token response.
It can be a string of any content that you wish. A randomly generated unique value is typically
used for preventing cross-site request forgery attacks. The state is also used to encode information
about the user's state in the app before the authentication request occurred, such as the page or view
they were on. Defaults to None.
Returns:
str: Url for OAuth 2.0.
"""
params = {
"client_id": self.client_id,
"redirect_uri": redirect_uri,
"scope": " ".join(scope),
"response_type": "code",
"response_mode": "query",
}
if state:
params["state"] = state
response = self.AUTHORITY_URL + self.account_type + self.AUTH_ENDPOINT + urlencode(params)
return response
def exchange_code(self, redirect_uri: str, code: str) -> Response:
"""Exchanges an oauth code for an user token.
Your app uses the authorization code received in the previous step to request an access token by sending a POST
request to the /token endpoint.
https://docs.microsoft.com/en-us/graph/auth-v2-user#3-get-a-token
Args:
redirect_uri (str): The redirect_uri of your app, where authentication responses can be sent and received by
your app. It must exactly match one of the redirect_uris you registered in the app registration portal.
code (str): The authorization_code that you acquired in the first leg of the flow.
Returns:
Response: Microsoft Graph Response.
"""
data = {
"client_id": self.client_id,
"redirect_uri": redirect_uri,
"client_secret": self.client_secret,
"code": code,
"grant_type": "authorization_code",
}
response = requests.post(self.AUTHORITY_URL + self.account_type + self.TOKEN_ENDPOINT, data=data)
return self._parse(response)
def refresh_token(self, redirect_uri: str, refresh_token: str) -> Response:
"""Exchanges a refresh token for an user token.
Access tokens are short lived, and you must refresh them after they expire to continue accessing resources.
You can do so by submitting another POST request to the /token endpoint, this time providing the refresh_token
instead of the code.
https://docs.microsoft.com/en-us/graph/auth-v2-user#5-use-the-refresh-token-to-get-a-new-access-token
Args:
redirect_uri (str): The redirect_uri of your app, where authentication responses can be sent and received by
your app. It must exactly match one of the redirect_uris you registered in the app registration portal.
refresh_token (str): An OAuth 2.0 refresh token. Your app can use this token acquire additional access tokens
after the current access token expires. Refresh tokens are long-lived, and can be used to retain access
to resources for extended periods of time.
Returns:
Response: Microsoft Graph Response.
"""
data = {
"client_id": self.client_id,
"redirect_uri": redirect_uri,
"client_secret": self.client_secret,
"refresh_token": refresh_token,
"grant_type": "refresh_token",
}
response = requests.post(self.AUTHORITY_URL + self.account_type + self.TOKEN_ENDPOINT, data=data)
return self._parse(response)
def set_token(self, token: dict) -> None:
"""Sets the User token for its use in this library.
Args:
token (dict): User token data.
"""
self.token = token
def set_workbook_session_id(self, workbook_session_id: dict) -> None:
"""Sets the Workbook Session Id token for its use in this library.
Args:
token (dict): Workbook Session ID.
"""
self.workbook_session_id = workbook_session_id
def get_next(self, response: Response) -> Optional[Response]:
"""Retrieves the next page for the argument response if any. This allows to perform a loop in case you
want to paginate the response yourself.
Args:
response (Response): Graph API Response.
Returns:
Optional[Response]: Graph API Response if available, None otherwise
"""
if not isinstance(response.data, dict):
return None
if "@odata.nextLink" not in response.data:
return None
return self._do_get(response.data["@odata.nextLink"])
def _paginate_response(self, response: Response) -> Response:
"""Some queries against Microsoft Graph return multiple pages of data either due to server-side paging or due to
the use of the $top query parameter to specifically limit the page size in a request. When a result set spans
multiple pages, Microsoft Graph returns an @odata.nextLink property in the response that contains a URL to the
next page of results.
https://docs.microsoft.com/en-us/graph/paging?context=graph%2Fapi%2F1.0&view=graph-rest-1.0
Args:
response (Response): Graph API Response.
Returns:
Response: Graph API Response.
"""
if not isinstance(response.data, dict) or "value" not in response.data:
return response
# Copy data to avoid side effects
data = list(response.data["value"])
while "@odata.nextLink" in response.data:
response = self.get_next(response)
if isinstance(response.data, dict) and "value" in response.data:
data.extend(response.data["value"])
response.data["value"] = data
return response
def _get(self, url, **kwargs) -> Response:
response = self._do_get(url, **kwargs)
if self.paginate:
return self._paginate_response(response)
return response
def _do_get(self, url, **kwargs) -> Response:
return self._request("GET", url, **kwargs)
def _post(self, url, **kwargs):
return self._request("POST", url, **kwargs)
def _put(self, url, **kwargs):
return self._request("PUT", url, **kwargs)
def _patch(self, url, **kwargs):
return self._request("PATCH", url, **kwargs)
def _delete(self, url, **kwargs):
return self._request("DELETE", url, **kwargs)
def _request(self, method, url, headers=None, **kwargs) -> Response:
_headers = {
"Accept": "application/json",
}
if headers:
_headers.update(headers)
_headers["Authorization"] = "Bearer " + self.token["access_token"]
if self.requests_hooks:
kwargs.update({"hooks": self.requests_hooks})
if "Content-Type" not in _headers:
_headers["Content-Type"] = "application/json"
return self._parse(requests.request(method, url, headers=_headers, **kwargs))
def _parse(self, response) -> Response:
status_code = response.status_code
r = Response(original=response)
if status_code < 299:
return r
elif status_code == 400:
raise exceptions.BadRequest(r.data)
elif status_code == 401:
raise exceptions.Unauthorized(r.data)
elif status_code == 403:
raise exceptions.Forbidden(r.data)
elif status_code == 404:
raise exceptions.NotFound(r.data)
elif status_code == 405:
raise exceptions.MethodNotAllowed(r.data)
elif status_code == 406:
raise exceptions.NotAcceptable(r.data)
elif status_code == 409:
raise exceptions.Conflict(r.data)
elif status_code == 410:
raise exceptions.Gone(r.data)
elif status_code == 411:
raise exceptions.LengthRequired(r.data)
elif status_code == 412:
raise exceptions.PreconditionFailed(r.data)
elif status_code == 413:
raise exceptions.RequestEntityTooLarge(r.data)
elif status_code == 415:
raise exceptions.UnsupportedMediaType(r.data)
elif status_code == 416:
raise exceptions.RequestedRangeNotSatisfiable(r.data)
elif status_code == 422:
raise exceptions.UnprocessableEntity(r.data)
elif status_code == 429:
raise exceptions.TooManyRequests(r.data)
elif status_code == 500:
raise exceptions.InternalServerError(r.data)
elif status_code == 501:
raise exceptions.NotImplemented(r.data)
elif status_code == 503:
raise exceptions.ServiceUnavailable(r.data)
elif status_code == 504:
raise exceptions.GatewayTimeout(r.data)
elif status_code == 507:
raise exceptions.InsufficientStorage(r.data)
elif status_code == 509:
raise exceptions.BandwidthLimitExceeded(r.data)
else:
if r["error"]["innerError"]["code"] == "lockMismatch":
# File is currently locked due to being open in the web browser
# while attempting to reupload a new version to the drive.
# Thus temporarily unavailable.
raise exceptions.ServiceUnavailable(r.data)
raise exceptions.UnknownError(r.data)