Add additional functionality to subscriptions

O365/subscriptions.py

@@ -1,7 +1,7 @@
 import datetime as dt
 from typing import Iterable, Mapping, Optional, Union
 
-from .utils import ApiComponent
+from .utils import ApiComponent, NEXT_LINK_KEYWORD, Pagination
 
 
 class Subscriptions(ApiComponent):
@@ -76,6 +76,27 @@ def _stringify_change_type(change_type: Union[str, Iterable[str]]) -> str:
             raise ValueError("change_type must contain at least one value.")
         return value
 
+    def get_subscription(
+        self,
+        subscription_id: str,
+        *,
+        params: Optional[Mapping[str, object]] = None,
+        **request_kwargs,
+    ) -> Optional[dict]:
+        """Retrieve a single webhook subscription by id."""
+        if not subscription_id:
+            raise ValueError("subscription_id must be provided.")
+        if params is not None and not isinstance(params, Mapping):
+            raise ValueError("params must be a mapping if provided.")
+
+        url = self._build_subscription_url(subscription_id)
+        response = self.con.get(url, params=params, **request_kwargs)
+
+        if not response:
+            return None
+
+        return response.json()
+
     def create_subscription(
         self,
         notification_url: str,
@@ -146,6 +167,38 @@ def create_subscription(
 
         return response.json()
 
+    def list_subscriptions(
+        self,
+        *,
+        limit: Optional[int] = None,
+        **request_kwargs,
+    ) -> Union[Iterable[dict], Pagination]:
+        """List webhook subscriptions visible to the current app/context."""
+        if limit is not None and limit <= 0:
+            raise ValueError("limit must be a positive integer.")
+
+        url = self._build_subscription_url()
+        response = self.con.get(url, **request_kwargs)
+        if not response:
+            return iter(())
+
+        data = response.json()
+        subscriptions = data.get("value", [])
+        next_link = data.get(NEXT_LINK_KEYWORD)
+
+        if next_link:
+            return Pagination(
+                parent=self,
+                data=subscriptions,
+                next_link=next_link,
+                limit=limit,
+            )
+
+        if limit is not None:
+            return subscriptions[:limit]
+
+        return subscriptions
+
     def renew_subscription(
         self,
         subscription_id: str,
@@ -175,6 +228,43 @@ def renew_subscription(
 
         return response.json()
 
+    def update_subscription(
+        self,
+        subscription_id: str,
+        *,
+        notification_url: Optional[str] = None,
+        expiration_datetime: Optional[dt.datetime] = None,
+        expiration_minutes: Optional[int] = None,
+        **request_kwargs,
+    ) -> Optional[dict]:
+        """Update subscription fields (expiration and/or notification URL)."""
+        if not subscription_id:
+            raise ValueError("subscription_id must be provided.")
+
+        payload = {}
+
+        if expiration_datetime is not None or expiration_minutes is not None:
+            payload[self._cc("expiration_date_time")] = self._format_subscription_expiration(
+                expiration_datetime=expiration_datetime,
+                expiration_minutes=expiration_minutes,
+            )
+
+        if notification_url is not None:
+            if not notification_url:
+                raise ValueError("notification_url, if provided, cannot be empty.")
+            payload[self._cc("notification_url")] = notification_url
+
+        if not payload:
+            raise ValueError("At least one of expiration or notification_url must be provided.")
+
+        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,

examples/subscriptions_example.py

@@ -3,16 +3,18 @@
 
 Quickstart for this example:
 1) Run Flask locally withg the following command:
-    - flask --app examples/subscription_account_webhook.py run --debug
+    - flask --app examples/subscriptions_example.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: 
+5) To list subscriptions, follow the example request below:
+    - http://<your-tunnel-host>/subscriptions/list
+6) 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:
+7) 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.
 """
@@ -38,7 +40,7 @@
             ])
 
 RESOURCE = "/me/mailFolders('inbox')/messages"
-DEFAULT_EXPIRATION_MINUTES = 55  # Graph requires renewals before the limit
+DEFAULT_EXPIRATION_MINUTES = 10069 # Maximum expiration is 10,070 in the future.
 
 app = Flask(__name__)
 
@@ -63,7 +65,7 @@ def create_subscription():
     client_state = request.args.get("client_state")
     resource = request.args.get("resource", RESOURCE)
 
-    subscription = account.subscriptions.create_subscription(
+    subscription = account.subscriptions().create_subscription(
         notification_url=notification_url,
         resource=resource,
         change_type="created",
@@ -73,10 +75,26 @@ def create_subscription():
     return jsonify(subscription), 201
 
 
+@app.get("/subscriptions/list")
+def list_subscriptions():
+    limit_raw = request.args.get("limit")
+    limit = None
+    if limit_raw is not None:
+        try:
+            limit = int(limit_raw)
+        except ValueError:
+            abort(400, description="limit must be an integer")
+        if limit <= 0:
+            abort(400, description="limit must be a positive integer")
+
+    subscriptions = account.subscriptions().list_subscriptions(limit=limit)
+    return jsonify(list(subscriptions)), 200
+
+
 @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(
+    updated = account.subscriptions().renew_subscription(
         subscription_id,
         expiration_minutes=expiration_minutes,
     )
@@ -85,7 +103,7 @@ def renew_subscription(subscription_id: str):
 
 @app.get("/subscriptions/<subscription_id>/delete")
 def delete_subscription(subscription_id: str):
-    deleted = account.subscriptions.delete_subscription(subscription_id)
+    deleted = account.subscriptions().delete_subscription(subscription_id)
     if not deleted:
         abort(404, description="Subscription not found")
     return ("", 204)