Add generic notification subscriptions through webhooks

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/mailbox.py

@@ -1069,5 +1069,4 @@ def get_settings(self):
 
         return self.mailbox_settings_constructor(
             parent=self, **{self._cloud_data_key: data}
-        )
-
+        )

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)

docs/latest/api/account.html

@@ -52,6 +52,7 @@
 <li class="toctree-l1 current"><a class="reference internal" href="../api.html">O365 API</a><ul class="current">
 <li class="toctree-l2 current"><a class="current reference internal" href="#">Account</a><ul>
 <li class="toctree-l3"><a class="reference internal" href="#O365.account.Account"><code class="docutils literal notranslate"><span class="pre">Account</span></code></a><ul>
+<li class="toctree-l4"><a class="reference internal" href="#O365.account.Account.subscriptions_constructor"><code class="docutils literal notranslate"><span class="pre">Account.subscriptions_constructor</span></code></a></li>
 <li class="toctree-l4"><a class="reference internal" href="#O365.account.Account.__init__"><code class="docutils literal notranslate"><span class="pre">Account.__init__()</span></code></a></li>
 <li class="toctree-l4"><a class="reference internal" href="#O365.account.Account.address_book"><code class="docutils literal notranslate"><span class="pre">Account.address_book()</span></code></a></li>
 <li class="toctree-l4"><a class="reference internal" href="#O365.account.Account.authenticate"><code class="docutils literal notranslate"><span class="pre">Account.authenticate()</span></code></a></li>
@@ -131,6 +132,12 @@ <h1>Account<a class="headerlink" href="#account" title="Link to this heading">
 <dt class="sig sig-object py" id="O365.account.Account">
 <em class="property"><span class="k"><span class="pre">class</span></span><span class="w"> </span></em><span class="sig-prename descclassname"><span class="pre">O365.account.</span></span><span class="sig-name descname"><span class="pre">Account</span></span><span class="sig-paren">(</span><em class="sig-param"><span class="n"><span class="pre">credentials</span></span><span class="p"><span class="pre">:</span></span><span class="w"> </span><span class="n"><span class="pre">Tuple</span><span class="p"><span class="pre">[</span></span><span class="pre">str</span><span class="p"><span class="pre">,</span></span><span class="w"> </span><span class="pre">str</span><span class="p"><span class="pre">]</span></span></span></em>, <em class="sig-param"><span class="keyword-only-separator o"><abbr title="Keyword-only parameters separator (PEP 3102)"><span class="pre">*</span></abbr></span></em>, <em class="sig-param"><span class="n"><span class="pre">username</span></span><span class="p"><span class="pre">:</span></span><span class="w"> </span><span class="n"><span class="pre">str</span><span class="w"> </span><span class="p"><span class="pre">|</span></span><span class="w"> </span><span class="pre">None</span></span><span class="w"> </span><span class="o"><span class="pre">=</span></span><span class="w"> </span><span class="default_value"><span class="pre">None</span></span></em>, <em class="sig-param"><span class="n"><span class="pre">protocol</span></span><span class="p"><span class="pre">:</span></span><span class="w"> </span><span class="n"><a class="reference internal" href="connection.html#O365.connection.Protocol" title="O365.connection.Protocol"><span class="pre">Protocol</span></a><span class="w"> </span><span class="p"><span class="pre">|</span></span><span class="w"> </span><span class="pre">None</span></span><span class="w"> </span><span class="o"><span class="pre">=</span></span><span class="w"> </span><span class="default_value"><span class="pre">None</span></span></em>, <em class="sig-param"><span class="n"><span class="pre">main_resource</span></span><span class="p"><span class="pre">:</span></span><span class="w"> </span><span class="n"><span class="pre">str</span><span class="w"> </span><span class="p"><span class="pre">|</span></span><span class="w"> </span><span class="pre">None</span></span><span class="w"> </span><span class="o"><span class="pre">=</span></span><span class="w"> </span><span class="default_value"><span class="pre">None</span></span></em>, <em class="sig-param"><span class="o"><span class="pre">**</span></span><span class="n"><span class="pre">kwargs</span></span></em><span class="sig-paren">)</span><a class="reference internal" href="../_modules/O365/account.html#Account"><span class="viewcode-link"><span class="pre">[source]</span></span></a><a class="headerlink" href="#O365.account.Account" title="Link to this definition"></a></dt>
 <dd><p>Bases: <code class="xref py py-class docutils literal notranslate"><span class="pre">object</span></code></p>
+<dl class="py attribute">
+<dt class="sig sig-object py" id="O365.account.Account.subscriptions_constructor">
+<span class="sig-name descname"><span class="pre">subscriptions_constructor</span></span><a class="headerlink" href="#O365.account.Account.subscriptions_constructor" title="Link to this definition"></a></dt>
+<dd><p>alias of <code class="xref py py-class docutils literal notranslate"><span class="pre">Subscriptions</span></code></p>
+</dd></dl>
+
 <dl class="py method">
 <dt class="sig sig-object py" id="O365.account.Account.__init__">
 <span class="sig-name descname"><span class="pre">__init__</span></span><span class="sig-paren">(</span><em class="sig-param"><span class="n"><span class="pre">credentials</span></span><span class="p"><span class="pre">:</span></span><span class="w"> </span><span class="n"><span class="pre">Tuple</span><span class="p"><span class="pre">[</span></span><span class="pre">str</span><span class="p"><span class="pre">,</span></span><span class="w"> </span><span class="pre">str</span><span class="p"><span class="pre">]</span></span></span></em>, <em class="sig-param"><span class="keyword-only-separator o"><abbr title="Keyword-only parameters separator (PEP 3102)"><span class="pre">*</span></abbr></span></em>, <em class="sig-param"><span class="n"><span class="pre">username</span></span><span class="p"><span class="pre">:</span></span><span class="w"> </span><span class="n"><span class="pre">str</span><span class="w"> </span><span class="p"><span class="pre">|</span></span><span class="w"> </span><span class="pre">None</span></span><span class="w"> </span><span class="o"><span class="pre">=</span></span><span class="w"> </span><span class="default_value"><span class="pre">None</span></span></em>, <em class="sig-param"><span class="n"><span class="pre">protocol</span></span><span class="p"><span class="pre">:</span></span><span class="w"> </span><span class="n"><a class="reference internal" href="connection.html#O365.connection.Protocol" title="O365.connection.Protocol"><span class="pre">Protocol</span></a><span class="w"> </span><span class="p"><span class="pre">|</span></span><span class="w"> </span><span class="pre">None</span></span><span class="w"> </span><span class="o"><span class="pre">=</span></span><span class="w"> </span><span class="default_value"><span class="pre">None</span></span></em>, <em class="sig-param"><span class="n"><span class="pre">main_resource</span></span><span class="p"><span class="pre">:</span></span><span class="w"> </span><span class="n"><span class="pre">str</span><span class="w"> </span><span class="p"><span class="pre">|</span></span><span class="w"> </span><span class="pre">None</span></span><span class="w"> </span><span class="o"><span class="pre">=</span></span><span class="w"> </span><span class="default_value"><span class="pre">None</span></span></em>, <em class="sig-param"><span class="o"><span class="pre">**</span></span><span class="n"><span class="pre">kwargs</span></span></em><span class="sig-paren">)</span><a class="reference internal" href="../_modules/O365/account.html#Account.__init__"><span class="viewcode-link"><span class="pre">[source]</span></span></a><a class="headerlink" href="#O365.account.Account.__init__" title="Link to this definition"></a></dt>

docs/latest/genindex.html

@@ -3288,10 +3288,10 @@ <h2 id="S">S</h2>
 </li>
       <li><a href="api/excel.html#O365.excel.RangeFormat.set_borders">set_borders() (O365.excel.RangeFormat method)</a>
 </li>
-  </ul></td>
-  <td style="width: 33%; vertical-align: top;"><ul>
       <li><a href="api/message.html#O365.message.MessageFlag.set_completed">set_completed() (O365.message.MessageFlag method)</a>
 </li>
+  </ul></td>
+  <td style="width: 33%; vertical-align: top;"><ul>
       <li><a href="api/calendar.html#O365.calendar.EventRecurrence.set_daily">set_daily() (O365.calendar.EventRecurrence method)</a>
 </li>
       <li><a href="api/mailbox.html#O365.mailbox.MailBox.set_disable_reply">set_disable_reply() (O365.mailbox.MailBox method)</a>
@@ -3438,6 +3438,8 @@ <h2 id="S">S</h2>
         <li><a href="api/teams.html#O365.teams.ChatMessage.subject">(O365.teams.ChatMessage attribute)</a>
 </li>
       </ul></li>
+      <li><a href="api/account.html#O365.account.Account.subscriptions_constructor">subscriptions_constructor (O365.account.Account attribute)</a>
+</li>
       <li><a href="api/teams.html#O365.teams.ChatMessage.summary">summary (O365.teams.ChatMessage attribute)</a>
 </li>
       <li><a href="api/address_book.html#O365.address_book.Contact.surname">surname (O365.address_book.Contact property)</a>