Source code for nc_py_api.ex_app.events_listener

"""Nextcloud API for registering Events listeners for ExApps."""

import dataclasses

from .._exceptions import NextcloudExceptionNotFound
from .._misc import require_capabilities
from .._session import AsyncNcSessionApp, NcSessionApp

_EP_SUFFIX: str = "events_listener"


[docs] @dataclasses.dataclass class EventsListener: """EventsListener description.""" def __init__(self, raw_data: dict): self._raw_data = raw_data @property def event_type(self) -> str: """Main type of event, e.g. ``node_event``.""" return self._raw_data["event_type"] @property def event_subtypes(self) -> str: """Subtypes for which fire event, e.g. ``NodeCreatedEvent``, ``NodeDeletedEvent``.""" return self._raw_data["event_subtypes"] @property def action_handler(self) -> str: """Relative ExApp url which will be called by Nextcloud.""" return self._raw_data["action_handler"] def __repr__(self): return f"<{self.__class__.__name__} event_type={self.event_type}, handler={self.action_handler}>"
[docs] class EventsListenerAPI: """API for registering Events listeners, avalaible as **nc.events_handler.<method>**.""" def __init__(self, session: NcSessionApp): self._session = session
[docs] def register( self, event_type: str, callback_url: str, event_subtypes: list[str] | None = None, ) -> None: """Registers or edits the events listener.""" if event_subtypes is None: event_subtypes = [] require_capabilities("app_api", self._session.capabilities) params = { "eventType": event_type, "actionHandler": callback_url, "eventSubtypes": event_subtypes, } self._session.ocs("POST", f"{self._session.ae_url}/{_EP_SUFFIX}", json=params)
[docs] def unregister(self, event_type: str, not_fail=True) -> None: """Removes the events listener.""" require_capabilities("app_api", self._session.capabilities) try: self._session.ocs( "DELETE", f"{self._session.ae_url}/{_EP_SUFFIX}", params={"eventType": event_type}, ) except NextcloudExceptionNotFound as e: if not not_fail: raise e from None
[docs] def get_entry(self, event_type: str) -> EventsListener | None: """Get information about the event listener.""" require_capabilities("app_api", self._session.capabilities) try: return EventsListener( self._session.ocs( "GET", f"{self._session.ae_url}/{_EP_SUFFIX}", params={"eventType": event_type}, ) ) except NextcloudExceptionNotFound: return None
class AsyncEventsListenerAPI: """API for registering Events listeners, avalaible as **nc.events_handler.<method>**.""" def __init__(self, session: AsyncNcSessionApp): self._session = session async def register( self, event_type: str, callback_url: str, event_subtypes: list[str] | None = None, ) -> None: """Registers or edits the events listener.""" if event_subtypes is None: event_subtypes = [] require_capabilities("app_api", await self._session.capabilities) params = { "eventType": event_type, "actionHandler": callback_url, "eventSubtypes": event_subtypes, } await self._session.ocs("POST", f"{self._session.ae_url}/{_EP_SUFFIX}", json=params) async def unregister(self, event_type: str, not_fail=True) -> None: """Removes the events listener.""" require_capabilities("app_api", await self._session.capabilities) try: await self._session.ocs( "DELETE", f"{self._session.ae_url}/{_EP_SUFFIX}", params={"eventType": event_type}, ) except NextcloudExceptionNotFound as e: if not not_fail: raise e from None async def get_entry(self, event_type: str) -> EventsListener | None: """Get information about the event listener.""" require_capabilities("app_api", await self._session.capabilities) try: return EventsListener( await self._session.ocs( "GET", f"{self._session.ae_url}/{_EP_SUFFIX}", params={"eventType": event_type}, ) ) except NextcloudExceptionNotFound: return None