Repository navigation
Expand file tree
/
Copy pathwebhooks.py
More file actions
88 lines (69 loc) · 3.24 KB
/
Copy pathwebhooks.py
File metadata and controls
88 lines (69 loc) · 3.24 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
from datetime import datetime
from microsoftgraph.decorators import token_required
from microsoftgraph.response import Response
from microsoftgraph.utils import format_time
class Webhooks(object):
def __init__(self, client) -> None:
"""Set up notifications for changes in user data.
https://docs.microsoft.com/en-us/graph/webhooks?view=graph-rest-1.0
Args:
client (Client): Library Client.
"""
self._client = client
@token_required
def create_subscription(
self,
change_type: str,
notification_url: str,
resource: str,
expiration_datetime: datetime,
client_state: str = None,
**kwargs,
) -> Response:
"""Creates a subscription to start receiving notifications for a resource.
https://docs.microsoft.com/en-us/graph/webhooks#creating-a-subscription
Args:
change_type (str): The event type that caused the notification. For example, created on mail receive, or
updated on marking a message read.
notification_url (str): Url to receive notifications.
resource (str): The URI of the resource relative to https://graph.microsoft.com.
expiration_datetime (datetime): The expiration time for the subscription.
client_state (str, optional): The clientState property specified in the subscription request. Defaults to None.
Returns:
Response: Microsoft Graph Response.
"""
if isinstance(expiration_datetime, datetime):
expiration_datetime = format_time(expiration_datetime, is_webhook=True)
data = {
"changeType": change_type,
"notificationUrl": notification_url,
"resource": resource,
"expirationDateTime": expiration_datetime,
"clientState": client_state,
}
data.update(kwargs)
return self._client._post(self._client.base_url + "subscriptions", json=data)
@token_required
def renew_subscription(self, subscription_id: str, expiration_datetime: datetime) -> Response:
"""Renews a subscription to keep receiving notifications for a resource.
https://docs.microsoft.com/en-us/graph/webhooks#renewing-a-subscription
Args:
subscription_id (str): Subscription ID.
expiration_datetime (datetime): Expiration date.
Returns:
Response: Microsoft Graph Response.
"""
if isinstance(expiration_datetime, datetime):
expiration_datetime = format_time(expiration_datetime, is_webhook=True)
data = {"expirationDateTime": expiration_datetime}
return self._client._patch(self._client.base_url + "subscriptions/{}".format(subscription_id), json=data)
@token_required
def delete_subscription(self, subscription_id: str) -> Response:
"""Deletes a subscription to stop receiving notifications for a resource.
https://docs.microsoft.com/en-us/graph/webhooks#deleting-a-subscription
Args:
subscription_id (str): Subscription ID.
Returns:
Response: Microsoft Graph Response.
"""
return self._client._delete(self._client.base_url + "subscriptions/{}".format(subscription_id))