feat: Events Listeners registration API (#249)

API to register Events Listeners for NextcloudApp

Reference: nextcloud/app_api#259

---------

Signed-off-by: Alexander Piskun <[email protected]>

Author: bigcat88

CHANGELOG.md

@@ -2,11 +2,12 @@
 
 All notable changes to this project will be documented in this file.
 
-## [0.13.0 - 2024-04-xx]
+## [0.13.0 - 2024-04-28]
 
 ### Added
 
-- `occ` commands registration API(AppAPI 2.5.0+). #24
+- NextcloudApp: `occ` commands registration API(AppAPI 2.5.0+). #247
+- NextcloudApp: `Nodes` events listener registration API(AppAPI 2.5.0+). #249
 
 ## [0.12.1 - 2024-04-05]
 

docs/reference/ExApp.rst

@@ -87,6 +87,12 @@ UI methods should be accessed with the help of :class:`~nc_py_api.nextcloud.Next
 .. autoclass:: nc_py_api.ex_app.providers.translations._TranslationsProviderAPI
     :members:
 
+.. autoclass:: nc_py_api.ex_app.events_listener.EventsListener
+    :members:
+
+.. autoclass:: nc_py_api.ex_app.events_listener.EventsListenerAPI
+    :members:
+
 .. autoclass:: nc_py_api.ex_app.occ_commands.OccCommand
     :members:
 

nc_py_api/ex_app/events_listener.py

@@ -0,0 +1,137 @@
+"""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"
+
+
+@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}>"
+
+
+class EventsListenerAPI:
+    """API for registering Events listeners, avalaible as **nc.events_handler.<method>**."""
+
+    def __init__(self, session: NcSessionApp):
+        self._session = session
+
+    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)
+
+    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
+
+    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

nc_py_api/nextcloud.py

@@ -30,6 +30,7 @@
 from .apps import _AppsAPI, _AsyncAppsAPI
 from .calendar import _CalendarAPI
 from .ex_app.defs import LogLvl
+from .ex_app.events_listener import AsyncEventsListenerAPI, EventsListenerAPI
 from .ex_app.occ_commands import AsyncOccCommandsAPI, OccCommandsAPI
 from .ex_app.providers.providers import AsyncProvidersApi, ProvidersApi
 from .ex_app.ui.ui import AsyncUiApi, UiApi
@@ -305,8 +306,10 @@ class NextcloudApp(_NextcloudBasic):
     """Nextcloud UI API for ExApps"""
     providers: ProvidersApi
     """API for registering providers for Nextcloud"""
+    events_listener: EventsListenerAPI
+    """API for registering Events listeners for ExApps"""
     occ_commands: OccCommandsAPI
-    """API for registering OCC command from ExApp"""
+    """API for registering OCC command for ExApps"""
 
     def __init__(self, **kwargs):
         """The parameters will be taken from the environment.
@@ -319,6 +322,7 @@ def __init__(self, **kwargs):
         self.preferences_ex = PreferencesExAPI(self._session)
         self.ui = UiApi(self._session)
         self.providers = ProvidersApi(self._session)
+        self.events_listener = EventsListenerAPI(self._session)
         self.occ_commands = OccCommandsAPI(self._session)
 
     def log(self, log_lvl: LogLvl, content: str) -> None:
@@ -425,8 +429,10 @@ class AsyncNextcloudApp(_AsyncNextcloudBasic):
     """Nextcloud UI API for ExApps"""
     providers: AsyncProvidersApi
     """API for registering providers for Nextcloud"""
+    events_listener: AsyncEventsListenerAPI
+    """API for registering Events listeners for ExApps"""
     occ_commands: AsyncOccCommandsAPI
-    """API for registering OCC command from ExApp"""
+    """API for registering OCC command for ExApps"""
 
     def __init__(self, **kwargs):
         """The parameters will be taken from the environment.
@@ -439,6 +445,7 @@ def __init__(self, **kwargs):
         self.preferences_ex = AsyncPreferencesExAPI(self._session)
         self.ui = AsyncUiApi(self._session)
         self.providers = AsyncProvidersApi(self._session)
+        self.events_listener = AsyncEventsListenerAPI(self._session)
         self.occ_commands = AsyncOccCommandsAPI(self._session)
 
     async def log(self, log_lvl: LogLvl, content: str) -> None:

tests/actual_tests/events_listener_test.py

@@ -0,0 +1,52 @@
+import pytest
+
+from nc_py_api import NextcloudExceptionNotFound
+
+
+def test_events_registration(nc_app):
+    nc_app.events_listener.register(
+        "node_event",
+        "/some_url",
+    )
+    result = nc_app.events_listener.get_entry("node_event")
+    assert result.event_type == "node_event"
+    assert result.action_handler == "some_url"
+    assert result.event_subtypes == []
+    nc_app.events_listener.register(
+        "node_event", callback_url="/new_url", event_subtypes=["NodeCreatedEvent", "NodeRenamedEvent"]
+    )
+    result = nc_app.events_listener.get_entry("node_event")
+    assert result.event_type == "node_event"
+    assert result.action_handler == "new_url"
+    assert result.event_subtypes == ["NodeCreatedEvent", "NodeRenamedEvent"]
+    nc_app.events_listener.unregister(result.event_type)
+    with pytest.raises(NextcloudExceptionNotFound):
+        nc_app.events_listener.unregister(result.event_type, not_fail=False)
+    nc_app.events_listener.unregister(result.event_type)
+    assert nc_app.events_listener.get_entry(result.event_type) is None
+    assert str(result).find("event_type=") != -1
+
+
+@pytest.mark.asyncio(scope="session")
+async def test_events_registration_async(anc_app):
+    await anc_app.events_listener.register(
+        "node_event",
+        "/some_url",
+    )
+    result = await anc_app.events_listener.get_entry("node_event")
+    assert result.event_type == "node_event"
+    assert result.action_handler == "some_url"
+    assert result.event_subtypes == []
+    await anc_app.events_listener.register(
+        "node_event", callback_url="/new_url", event_subtypes=["NodeCreatedEvent", "NodeRenamedEvent"]
+    )
+    result = await anc_app.events_listener.get_entry("node_event")
+    assert result.event_type == "node_event"
+    assert result.action_handler == "new_url"
+    assert result.event_subtypes == ["NodeCreatedEvent", "NodeRenamedEvent"]
+    await anc_app.events_listener.unregister(result.event_type)
+    with pytest.raises(NextcloudExceptionNotFound):
+        await anc_app.events_listener.unregister(result.event_type, not_fail=False)
+    await anc_app.events_listener.unregister(result.event_type)
+    assert await anc_app.events_listener.get_entry(result.event_type) is None
+    assert str(result).find("event_type=") != -1