Source code for nc_py_api.webhooks

"""Nextcloud Webhooks API."""

import dataclasses

from ._misc import clear_from_params_empty  # , require_capabilities
from ._session import AppConfig, AsyncNcSessionBasic, NcSessionBasic


[docs] @dataclasses.dataclass class WebhookInfo: """Information about the Webhook.""" def __init__(self, raw_data: dict): self._raw_data = raw_data @property def webhook_id(self) -> int: """`ID` of the webhook.""" return self._raw_data["id"] @property def app_id(self) -> str: """`ID` of the ExApp that registered webhook.""" return self._raw_data["appId"] if self._raw_data["appId"] else "" @property def user_id(self) -> str: """`UserID` if webhook was registered in user context.""" return self._raw_data["userId"] if self._raw_data["userId"] else "" @property def http_method(self) -> str: """HTTP method used to call webhook.""" return self._raw_data["httpMethod"] @property def uri(self) -> str: """URL address that will be called for this webhook.""" return self._raw_data["uri"] @property def event(self) -> str: """Nextcloud PHP event that triggers this webhook.""" return self._raw_data["event"] @property def event_filter(self): """Mongo filter to apply to the serialized data to decide if firing.""" return self._raw_data["eventFilter"] @property def user_id_filter(self) -> str: """Currently unknown.""" return self._raw_data["userIdFilter"] @property def headers(self) -> dict: """Headers that should be added to request when calling webhook.""" return self._raw_data["headers"] if self._raw_data["headers"] else {} @property def auth_method(self) -> str: """Currently unknown.""" return self._raw_data["authMethod"] @property def auth_data(self) -> dict: """Currently unknown.""" return self._raw_data["authData"] if self._raw_data["authData"] else {} def __repr__(self): return f"<{self.__class__.__name__} id={self.webhook_id}, event={self.event}>"
[docs] class _WebhooksAPI: """The class provides the application management API on the Nextcloud server.""" _ep_base: str = "/ocs/v1.php/apps/webhook_listeners/api/v1/webhooks" def __init__(self, session: NcSessionBasic): self._session = session def get_list(self, uri_filter: str = "") -> list[WebhookInfo]: params = {"uri": uri_filter} if uri_filter else {} return [WebhookInfo(i) for i in self._session.ocs("GET", f"{self._ep_base}", params=params)] def get_entry(self, webhook_id: int) -> WebhookInfo: return WebhookInfo(self._session.ocs("GET", f"{self._ep_base}/{webhook_id}")) def register( self, http_method: str, uri: str, event: str, event_filter: dict | None = None, user_id_filter: str = "", headers: dict | None = None, auth_method: str = "none", auth_data: dict | None = None, ): params = { "httpMethod": http_method, "uri": uri, "event": event, "eventFilter": event_filter, "userIdFilter": user_id_filter, "headers": headers, "authMethod": auth_method, "authData": auth_data, } clear_from_params_empty(["eventFilter", "userIdFilter", "headers", "authMethod", "authData"], params) return WebhookInfo(self._session.ocs("POST", f"{self._ep_base}", json=params)) def update( self, webhook_id: int, http_method: str, uri: str, event: str, event_filter: dict | None = None, user_id_filter: str = "", headers: dict | None = None, auth_method: str = "none", auth_data: dict | None = None, ): params = { "id": webhook_id, "httpMethod": http_method, "uri": uri, "event": event, "eventFilter": event_filter, "userIdFilter": user_id_filter, "headers": headers, "authMethod": auth_method, "authData": auth_data, } clear_from_params_empty(["eventFilter", "userIdFilter", "headers", "authMethod", "authData"], params) return WebhookInfo(self._session.ocs("POST", f"{self._ep_base}/{webhook_id}", json=params)) def unregister(self, webhook_id: int) -> bool: return self._session.ocs("DELETE", f"{self._ep_base}/{webhook_id}") def unregister_all(self, appid: str = "") -> int: if not appid and isinstance(self._session.cfg, AppConfig): appid = self._session.cfg.app_name else: raise ValueError("The `appid` parameter cannot be empty for non-ExApp use.") return self._session.ocs("DELETE", f"{self._ep_base}/byappid/{appid}")
class _AsyncWebhooksAPI: """The class provides the async application management API on the Nextcloud server.""" _ep_base: str = "/ocs/v1.php/webhooks" def __init__(self, session: AsyncNcSessionBasic): self._session = session async def get_list(self, uri_filter: str = "") -> list[WebhookInfo]: params = {"uri": uri_filter} if uri_filter else {} return [WebhookInfo(i) for i in await self._session.ocs("GET", f"{self._ep_base}", params=params)] async def get_entry(self, webhook_id: int) -> WebhookInfo: return WebhookInfo(await self._session.ocs("GET", f"{self._ep_base}/{webhook_id}")) async def register( self, http_method: str, uri: str, event: str, event_filter: dict | None = None, user_id_filter: str = "", headers: dict | None = None, auth_method: str = "none", auth_data: dict | None = None, ): params = { "httpMethod": http_method, "uri": uri, "event": event, "eventFilter": event_filter, "userIdFilter": user_id_filter, "headers": headers, "authMethod": auth_method, "authData": auth_data, } clear_from_params_empty(["eventFilter", "userIdFilter", "headers", "authMethod", "authData"], params) return WebhookInfo(await self._session.ocs("POST", f"{self._ep_base}", json=params)) async def update( self, webhook_id: int, http_method: str, uri: str, event: str, event_filter: dict | None = None, user_id_filter: str = "", headers: dict | None = None, auth_method: str = "none", auth_data: dict | None = None, ): params = { "id": webhook_id, "httpMethod": http_method, "uri": uri, "event": event, "eventFilter": event_filter, "userIdFilter": user_id_filter, "headers": headers, "authMethod": auth_method, "authData": auth_data, } clear_from_params_empty(["eventFilter", "userIdFilter", "headers", "authMethod", "authData"], params) return WebhookInfo(await self._session.ocs("POST", f"{self._ep_base}/{webhook_id}", json=params)) async def unregister(self, webhook_id: int) -> bool: return await self._session.ocs("DELETE", f"{self._ep_base}/{webhook_id}") async def unregister_all(self, appid: str = "") -> int: if not appid and isinstance(self._session.cfg, AppConfig): appid = self._session.cfg.app_name else: raise ValueError("The `appid` parameter cannot be empty for non-ExApp use.") return await self._session.ocs("DELETE", f"{self._ep_base}/byappid/{appid}")