Added subscriptions to Account and example

Author: sdelgadoc

O365/account.py

@@ -2,16 +2,18 @@
 from typing import Callable, List, Optional, Tuple, Type
 
 from .connection import Connection, MSGraphProtocol, Protocol
+from .subscriptions import Subscriptions
 from .utils import ME_RESOURCE, consent_input_token
 
 
-class Account:
-    connection_constructor: Type = Connection  #: :meta private:
-
-    def __init__(self, credentials: Tuple[str, str], *,
-                 username: Optional[str] = None,
-                 protocol: Optional[Protocol] = None,
-                 main_resource: Optional[str] = None, **kwargs):
+class Account:
+    connection_constructor: Type = Connection  #: :meta private:
+    subscriptions_constructor: Type = Subscriptions
+
+    def __init__(self, credentials: Tuple[str, str], *,
+                 username: Optional[str] = None,
+                 protocol: Optional[Protocol] = None,
+                 main_resource: Optional[str] = None, **kwargs):
         """ Creates an object which is used to access resources related to the specified credentials.
 
         :param credentials: a tuple containing the client_id and client_secret
@@ -60,6 +62,7 @@ def __init__(self, credentials: Tuple[str, str], *,
         self.con = self.connection_constructor(credentials, **kwargs)
         #: The resource in use for the account. |br| **Type:** str
         self.main_resource: str = main_resource or self.protocol.default_resource
+        self.subscriptions = self.subscriptions_constructor(parent=self)
 
     def __repr__(self):
         if self.con.auth:

O365/subscriptions.py

@@ -0,0 +1,190 @@
+import datetime as dt
+from typing import Iterable, Mapping, Optional, Union
+
+from .utils import ApiComponent
+
+
+class Subscriptions(ApiComponent):
+    """Subscription operations for Microsoft Graph webhooks."""
+
+    _endpoints = {
+        "subscriptions": "/subscriptions",
+    }
+
+    def __init__(self, *, parent=None, con=None, **kwargs):
+        if parent and con:
+            raise ValueError("Need a parent or a connection but not both")
+        self.con = parent.con if parent else con
+
+        main_resource = kwargs.pop("main_resource", None) or (
+            getattr(parent, "main_resource", None) if parent else None
+        )
+
+        super().__init__(
+            protocol=parent.protocol if parent else kwargs.get("protocol"),
+            main_resource=main_resource,
+        )
+
+    def _build_subscription_url(self, subscription_id: Optional[str] = None) -> str:
+        """Build the Microsoft Graph subscriptions endpoint."""
+        endpoint = self._endpoints.get("subscriptions")
+        if endpoint is None:
+            raise ValueError("Subscriptions endpoint is not configured.")
+        base_url = self.protocol.service_url.rstrip("/")
+        if subscription_id:
+            return f"{base_url}{endpoint}/{subscription_id}"
+        return f"{base_url}{endpoint}"
+
+    @staticmethod
+    def _format_subscription_expiration(
+        expiration_datetime: Optional[dt.datetime] = None,
+        expiration_minutes: Optional[int] = None,
+    ) -> str:
+        """Return an ISO 8601 UTC expiration string as required by Graph webhooks."""
+        if expiration_datetime and expiration_minutes is not None:
+            raise ValueError(
+                "Provide either expiration_datetime or expiration_minutes, not both."
+            )
+        if expiration_datetime is None:
+            minutes = expiration_minutes if expiration_minutes is not None else 60
+            if minutes <= 0:
+                raise ValueError("expiration_minutes must be a positive integer.")
+            expiration_datetime = dt.datetime.now(dt.timezone.utc) + dt.timedelta(
+                minutes=minutes
+            )
+        else:
+            if expiration_datetime.tzinfo is None:
+                expiration_datetime = expiration_datetime.replace(tzinfo=dt.timezone.utc)
+            else:
+                expiration_datetime = expiration_datetime.astimezone(dt.timezone.utc)
+        return expiration_datetime.isoformat(timespec="microseconds").replace("+00:00", "Z")
+
+    @staticmethod
+    def _stringify_change_type(change_type: Union[str, Iterable[str]]) -> str:
+        """Normalize changeType into the comma-separated string Graph expects."""
+        if isinstance(change_type, str):
+            value = change_type.strip()
+        else:
+            try:
+                parts = [str(part).strip() for part in change_type]
+            except TypeError as exc:
+                raise ValueError(
+                    "change_type must be a string or an iterable of strings."
+                ) from exc
+            value = ",".join(part for part in parts if part)
+        if not value:
+            raise ValueError("change_type must contain at least one value.")
+        return value
+
+    def create_subscription(
+        self,
+        notification_url: str,
+        resource: Optional[str] = None,
+        change_type: Union[str, Iterable[str]] = "created",
+        *,
+        expiration_datetime: Optional[dt.datetime] = None,
+        expiration_minutes: Optional[int] = None,
+        client_state: Optional[str] = None,
+        include_resource_data: Optional[bool] = None,
+        encryption_certificate: Optional[str] = None,
+        encryption_certificate_id: Optional[str] = None,
+        lifecycle_notification_url: Optional[str] = None,
+        latest_supported_tls_version: Optional[str] = None,
+        additional_data: Optional[Mapping[str, object]] = None,
+        **request_kwargs,
+    ) -> Optional[dict]:
+        """Create a Microsoft Graph webhook subscription.
+
+        See Documentation.md for webhook setup requirements.
+        """
+        if not notification_url:
+            raise ValueError("notification_url must be provided.")
+
+        resource = resource or self.main_resource
+        if not resource:
+            raise ValueError("resource must be provided.")
+        if not resource.startswith("/"):
+            resource = f"/{resource}"
+
+        expiration_value = self._format_subscription_expiration(
+            expiration_datetime=expiration_datetime,
+            expiration_minutes=expiration_minutes,
+        )
+        change_type_value = self._stringify_change_type(change_type)
+
+        payload = {
+            self._cc("change_type"): change_type_value,
+            self._cc("notification_url"): notification_url,
+            self._cc("resource"): resource,
+            self._cc("expiration_date_time"): expiration_value,
+        }
+
+        if client_state is not None:
+            payload[self._cc("client_state")] = client_state
+        if include_resource_data is not None:
+            payload[self._cc("include_resource_data")] = include_resource_data
+        if encryption_certificate is not None:
+            payload[self._cc("encryption_certificate")] = encryption_certificate
+        if encryption_certificate_id is not None:
+            payload[self._cc("encryption_certificate_id")] = encryption_certificate_id
+        if lifecycle_notification_url is not None:
+            payload[self._cc("lifecycle_notification_url")] = lifecycle_notification_url
+        if latest_supported_tls_version is not None:
+            payload[
+                self._cc("latest_supported_tls_version")
+            ] = latest_supported_tls_version
+        if additional_data:
+            if not isinstance(additional_data, Mapping):
+                raise ValueError("additional_data must be a mapping if provided.")
+            payload.update({str(key): value for key, value in additional_data.items()})
+
+        url = self._build_subscription_url()
+        response = self.con.post(url, data=payload, **request_kwargs)
+
+        if not response:
+            return None
+
+        return response.json()
+
+    def renew_subscription(
+        self,
+        subscription_id: str,
+        *,
+        expiration_datetime: Optional[dt.datetime] = None,
+        expiration_minutes: Optional[int] = None,
+        **request_kwargs,
+    ) -> Optional[dict]:
+        """Renew an existing webhook subscription."""
+        if not subscription_id:
+            raise ValueError("subscription_id must be provided.")
+
+        expiration_value = self._format_subscription_expiration(
+            expiration_datetime=expiration_datetime,
+            expiration_minutes=expiration_minutes,
+        )
+
+        payload = {
+            self._cc("expiration_date_time"): expiration_value,
+        }
+
+        url = self._build_subscription_url(subscription_id)
+        response = self.con.patch(url, data=payload, **request_kwargs)
+
+        if not response:
+            return None
+
+        return response.json()
+
+    def delete_subscription(
+        self,
+        subscription_id: str,
+        **request_kwargs,
+    ) -> bool:
+        """Delete an existing webhook subscription."""
+        if not subscription_id:
+            raise ValueError("subscription_id must be provided.")
+
+        url = self._build_subscription_url(subscription_id)
+        response = self.con.delete(url, **request_kwargs)
+
+        return bool(response)

examples/subscriptions_example.py

@@ -0,0 +1,114 @@
+
+""" Example on how to use and setup webhooks
+
+Quickstart for this example:
+1) Run Flask locally withg the following command:
+    - flask --app examples/subscription_account_webhook.py run --debug
+2) Expose HTTPS via a tunnel to your localhost:5000:
+    - Free: pinggy (https://pinggy.io/) to get https://<subdomain>.pinggy.link -> http://localhost:5000
+    - Paid/free-tier: ngrok (https://ngrok.com/): ngrok http 5000, note the https URL.
+3) Use the tunnel HTTPS URL as notification_url pointing to /webhook, URL-encoded. 
+4) To create a subscription, follow the example request below:
+    - https://<your-tunnel-host>/subscriptions?notification_url=https%3A%2F%2F<your-tunnel-host>%2Fwebhook&client_state=abc123
+4) To renew a subscription, follow the example request below: 
+    - http://<your-tunnel-host>/subscriptions/<subscription_id>/renew?expiration_minutes=55
+5) To delete a subscription, follow the example request below:
+    - http://<your-tunnel-host>/subscriptions/<subscription_id>/delete
+Graph will call https://<your-tunnel-host>/webhook; this app echoes validationToken and returns 202 for notifications.
+"""
+
+from flask import Flask, abort, jsonify, request
+from O365 import Account
+
+CLIENT_ID = "YOUR CLIENT ID"
+CLIENT_SECRET = "YOUR CLIENT SECRET"
+credentials = (CLIENT_ID, CLIENT_SECRET)
+
+account = Account(credentials)
+# Pick the scopes that are relevant to you here
+account.authenticate(
+            scopes=[
+                "https://graph.microsoft.com/Mail.ReadWrite",
+                "https://graph.microsoft.com/Mail.Send",
+                "https://graph.microsoft.com/Calendars.ReadWrite",
+                "https://graph.microsoft.com/MailboxSettings.ReadWrite",
+                "https://graph.microsoft.com/User.Read",
+                "https://graph.microsoft.com/User.ReadBasic.All",
+                'offline_access'
+            ])
+
+RESOURCE = "/me/mailFolders('inbox')/messages"
+DEFAULT_EXPIRATION_MINUTES = 55  # Graph requires renewals before the limit
+
+app = Flask(__name__)
+
+
+def _int_arg(name: str, default: int) -> int:
+    raw = request.args.get(name)
+    if raw is None:
+        return default
+    try:
+        return int(raw)
+    except ValueError:
+        abort(400, description=f"{name} must be an integer")
+
+
+@app.get("/subscriptions")
+def create_subscription():
+    notification_url = request.args.get("notification_url")
+    if not notification_url:
+        abort(400, description="notification_url is required")
+
+    expiration_minutes = _int_arg("expiration_minutes", DEFAULT_EXPIRATION_MINUTES)
+    client_state = request.args.get("client_state")
+    resource = request.args.get("resource", RESOURCE)
+
+    subscription = account.subscriptions.create_subscription(
+        notification_url=notification_url,
+        resource=resource,
+        change_type="created",
+        expiration_minutes=expiration_minutes,
+        client_state=client_state,
+    )
+    return jsonify(subscription), 201
+
+
+@app.get("/subscriptions/<subscription_id>/renew")
+def renew_subscription(subscription_id: str):
+    expiration_minutes = _int_arg("expiration_minutes", DEFAULT_EXPIRATION_MINUTES)
+    updated = account.subscriptions.renew_subscription(
+        subscription_id,
+        expiration_minutes=expiration_minutes,
+    )
+    return jsonify(updated), 200
+
+
+@app.get("/subscriptions/<subscription_id>/delete")
+def delete_subscription(subscription_id: str):
+    deleted = account.subscriptions.delete_subscription(subscription_id)
+    if not deleted:
+        abort(404, description="Subscription not found")
+    return ("", 204)
+
+
+@app.post("/webhook")
+def webhook_handler():
+    """Handle Microsoft Graph webhook calls.
+
+    - During subscription validation, Graph sends POST with ?validationToken=... .
+      We must echo the token as plain text within 10 seconds.
+    - For change notifications, Graph posts JSON; we just log/ack.
+    """
+    validation_token = request.args.get("validationToken")
+    if validation_token:
+        # Echo back token exactly as plain text with HTTP 200.
+        return validation_token, 200, {"Content-Type": "text/plain"}
+
+    # Change notifications: inspect or log as needed.
+    payload = request.get_json(silent=True) or {}
+    print("Received notification payload:", payload)
+    return ("", 202)
+
+
+if __name__ == "__main__":
+    app.run(debug=True, ssl_context=("examples/cert.pem", "examples/key.pem"))