feat: ExApps: occ-commands registration (#247)

API fore registering & unregistering OCC commands

Reference: cloud-py-api/app_api#272

Signed-off-by: Alexander Piskun <bigcat88@icloud.com>

CHANGELOG.md

@@ -2,6 +2,12 @@
 
 All notable changes to this project will be documented in this file.
 
+## [0.13.0 - 2024-04-xx]
+
+### Added
+
+- `occ` commands registration API(AppAPI 2.5.0+). #24
+
 ## [0.12.1 - 2024-04-05]
 
 ### Fixed

docs/reference/ExApp.rst

@@ -86,3 +86,9 @@ 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.occ_commands.OccCommand
+    :members:
+
+.. autoclass:: nc_py_api.ex_app.occ_commands.OccCommandsAPI
+    :members:

nc_py_api/ex_app/occ_commands.py

@@ -0,0 +1,153 @@
+"""Nextcloud API for registering OCC commands for ExApps."""
+
+import dataclasses
+
+from .._exceptions import NextcloudExceptionNotFound
+from .._misc import clear_from_params_empty, require_capabilities
+from .._session import AsyncNcSessionApp, NcSessionApp
+
+_EP_SUFFIX: str = "occ_command"
+
+
+@dataclasses.dataclass
+class OccCommand:
+    """OccCommand description."""
+
+    def __init__(self, raw_data: dict):
+        self._raw_data = raw_data
+
+    @property
+    def name(self) -> str:
+        """Unique ID for the command."""
+        return self._raw_data["name"]
+
+    @property
+    def description(self) -> str:
+        """Command description."""
+        return self._raw_data["description"]
+
+    @property
+    def hidden(self) -> bool:
+        """Flag determining ss command hidden or not."""
+        return self._raw_data["hidden"]
+
+    @property
+    def arguments(self) -> dict:
+        """Look at PHP Symfony framework for details."""
+        return self._raw_data["arguments"]
+
+    @property
+    def options(self) -> str:
+        """Look at PHP Symfony framework for details."""
+        return self._raw_data["options"]
+
+    @property
+    def usages(self) -> str:
+        """Look at PHP Symfony framework for details."""
+        return self._raw_data["usages"]
+
+    @property
+    def action_handler(self) -> str:
+        """Relative ExApp url which will be called by Nextcloud."""
+        return self._raw_data["execute_handler"]
+
+    def __repr__(self):
+        return f"<{self.__class__.__name__} name={self.name}, handler={self.action_handler}>"
+
+
+class OccCommandsAPI:
+    """API for registering OCC commands, avalaible as **nc.occ_command.<method>**."""
+
+    def __init__(self, session: NcSessionApp):
+        self._session = session
+
+    def register(
+        self,
+        name: str,
+        callback_url: str,
+        arguments: list | None = None,
+        options: list | None = None,
+        usages: list | None = None,
+        description: str = "",
+        hidden: bool = False,
+    ) -> None:
+        """Registers or edit the OCC command."""
+        require_capabilities("app_api", self._session.capabilities)
+        params = {
+            "name": name,
+            "description": description,
+            "arguments": arguments,
+            "hidden": hidden,
+            "options": options,
+            "usages": usages,
+            "execute_handler": callback_url,
+        }
+        clear_from_params_empty(["arguments", "options", "usages"], params)
+        self._session.ocs("POST", f"{self._session.ae_url}/{_EP_SUFFIX}", json=params)
+
+    def unregister(self, name: str, not_fail=True) -> None:
+        """Removes the OCC command."""
+        require_capabilities("app_api", self._session.capabilities)
+        try:
+            self._session.ocs("DELETE", f"{self._session.ae_url}/{_EP_SUFFIX}", params={"name": name})
+        except NextcloudExceptionNotFound as e:
+            if not not_fail:
+                raise e from None
+
+    def get_entry(self, name: str) -> OccCommand | None:
+        """Get information of the OCC command."""
+        require_capabilities("app_api", self._session.capabilities)
+        try:
+            return OccCommand(self._session.ocs("GET", f"{self._session.ae_url}/{_EP_SUFFIX}", params={"name": name}))
+        except NextcloudExceptionNotFound:
+            return None
+
+
+class AsyncOccCommandsAPI:
+    """Async API for registering OCC commands, avalaible as **nc.occ_command.<method>**."""
+
+    def __init__(self, session: AsyncNcSessionApp):
+        self._session = session
+
+    async def register(
+        self,
+        name: str,
+        callback_url: str,
+        arguments: list | None = None,
+        options: list | None = None,
+        usages: list | None = None,
+        description: str = "",
+        hidden: bool = False,
+    ) -> None:
+        """Registers or edit the OCC command."""
+        require_capabilities("app_api", await self._session.capabilities)
+        params = {
+            "name": name,
+            "description": description,
+            "arguments": arguments,
+            "hidden": hidden,
+            "options": options,
+            "usages": usages,
+            "execute_handler": callback_url,
+        }
+        clear_from_params_empty(["arguments", "options", "usages"], params)
+        await self._session.ocs("POST", f"{self._session.ae_url}/{_EP_SUFFIX}", json=params)
+
+    async def unregister(self, name: str, not_fail=True) -> None:
+        """Removes the OCC command."""
+        require_capabilities("app_api", await self._session.capabilities)
+        try:
+            await self._session.ocs("DELETE", f"{self._session.ae_url}/{_EP_SUFFIX}", params={"name": name})
+        except NextcloudExceptionNotFound as e:
+            if not not_fail:
+                raise e from None
+
+    async def get_entry(self, name: str) -> OccCommand | None:
+        """Get information of the OCC command."""
+        require_capabilities("app_api", await self._session.capabilities)
+        try:
+            return OccCommand(
+                await self._session.ocs("GET", f"{self._session.ae_url}/{_EP_SUFFIX}", params={"name": name})
+            )
+        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.occ_commands import AsyncOccCommandsAPI, OccCommandsAPI
 from .ex_app.providers.providers import AsyncProvidersApi, ProvidersApi
 from .ex_app.ui.ui import AsyncUiApi, UiApi
 from .files.files import AsyncFilesAPI, FilesAPI
@@ -304,6 +305,8 @@ class NextcloudApp(_NextcloudBasic):
     """Nextcloud UI API for ExApps"""
     providers: ProvidersApi
     """API for registering providers for Nextcloud"""
+    occ_commands: OccCommandsAPI
+    """API for registering OCC command from ExApp"""
 
     def __init__(self, **kwargs):
         """The parameters will be taken from the environment.
@@ -316,6 +319,7 @@ def __init__(self, **kwargs):
         self.preferences_ex = PreferencesExAPI(self._session)
         self.ui = UiApi(self._session)
         self.providers = ProvidersApi(self._session)
+        self.occ_commands = OccCommandsAPI(self._session)
 
     def log(self, log_lvl: LogLvl, content: str) -> None:
         """Writes log to the Nextcloud log file."""
@@ -421,6 +425,8 @@ class AsyncNextcloudApp(_AsyncNextcloudBasic):
     """Nextcloud UI API for ExApps"""
     providers: AsyncProvidersApi
     """API for registering providers for Nextcloud"""
+    occ_commands: AsyncOccCommandsAPI
+    """API for registering OCC command from ExApp"""
 
     def __init__(self, **kwargs):
         """The parameters will be taken from the environment.
@@ -433,6 +439,7 @@ def __init__(self, **kwargs):
         self.preferences_ex = AsyncPreferencesExAPI(self._session)
         self.ui = AsyncUiApi(self._session)
         self.providers = AsyncProvidersApi(self._session)
+        self.occ_commands = AsyncOccCommandsAPI(self._session)
 
     async def log(self, log_lvl: LogLvl, content: str) -> None:
         """Writes log to the Nextcloud log file."""

pyproject.toml

@@ -136,7 +136,7 @@ design.max-attributes = 8
 design.max-locals = 20
 design.max-branches = 16
 design.max-returns = 8
-design.max-args = 7
+design.max-args = 8
 basic.good-names = [
     "a", "b", "c", "d", "e", "f", "i", "j", "k", "r", "v",
     "ex", "_", "fp", "im", "nc", "ui",

tests/actual_tests/occ_commands_test.py

@@ -0,0 +1,124 @@
+import pytest
+
+from nc_py_api import NextcloudExceptionNotFound
+
+
+def test_occ_commands_registration(nc_app):
+    nc_app.occ_commands.register(
+        "test_occ_name",
+        "/some_url",
+    )
+    result = nc_app.occ_commands.get_entry("test_occ_name")
+    assert result.name == "test_occ_name"
+    assert result.description == ""
+    assert result.action_handler == "some_url"
+    assert result.hidden is False
+    assert result.usages == []
+    assert result.arguments == []
+    assert result.options == []
+    nc_app.occ_commands.register(
+        "test_occ_name2",
+        "some_url2",
+        description="desc",
+        arguments=[
+            {
+                "name": "argument_name",
+                "mode": "required",
+                "description": "Description of the argument",
+                "default": "default_value",
+            },
+        ],
+        options=[],
+    )
+    result2 = nc_app.occ_commands.get_entry("test_occ_name2")
+    assert result2.name == "test_occ_name2"
+    assert result2.description == "desc"
+    assert result2.action_handler == "some_url2"
+    assert result2.hidden is False
+    assert result2.usages == []
+    assert result2.arguments == [
+        {
+            "name": "argument_name",
+            "mode": "required",
+            "description": "Description of the argument",
+            "default": "default_value",
+        }
+    ]
+    assert result2.options == []
+    nc_app.occ_commands.register(
+        "test_occ_name",
+        description="new desc",
+        callback_url="/new_url",
+    )
+    result = nc_app.occ_commands.get_entry("test_occ_name")
+    assert result.name == "test_occ_name"
+    assert result.description == "new desc"
+    assert result.action_handler == "new_url"
+    nc_app.occ_commands.unregister(result.name)
+    with pytest.raises(NextcloudExceptionNotFound):
+        nc_app.occ_commands.unregister(result.name, not_fail=False)
+    nc_app.occ_commands.unregister(result.name)
+    nc_app.occ_commands.unregister(result2.name, not_fail=False)
+    assert nc_app.occ_commands.get_entry(result2.name) is None
+    assert str(result).find("name=") != -1
+
+
+@pytest.mark.asyncio(scope="session")
+async def test_occ_commands_registration_async(anc_app):
+    await anc_app.occ_commands.register(
+        "test_occ_name",
+        "/some_url",
+    )
+    result = await anc_app.occ_commands.get_entry("test_occ_name")
+    assert result.name == "test_occ_name"
+    assert result.description == ""
+    assert result.action_handler == "some_url"
+    assert result.hidden is False
+    assert result.usages == []
+    assert result.arguments == []
+    assert result.options == []
+    await anc_app.occ_commands.register(
+        "test_occ_name2",
+        "some_url2",
+        description="desc",
+        arguments=[
+            {
+                "name": "argument_name",
+                "mode": "required",
+                "description": "Description of the argument",
+                "default": "default_value",
+            },
+        ],
+        options=[],
+    )
+    result2 = await anc_app.occ_commands.get_entry("test_occ_name2")
+    assert result2.name == "test_occ_name2"
+    assert result2.description == "desc"
+    assert result2.action_handler == "some_url2"
+    assert result2.hidden is False
+    assert result2.usages == []
+    assert result2.arguments == [
+        {
+            "name": "argument_name",
+            "mode": "required",
+            "description": "Description of the argument",
+            "default": "default_value",
+        }
+    ]
+    assert result2.options == []
+    await anc_app.occ_commands.register(
+        "test_occ_name",
+        description="new desc",
+        callback_url="/new_url",
+    )
+    result = await anc_app.occ_commands.get_entry("test_occ_name")
+    assert result.name == "test_occ_name"
+    assert result.description == "new desc"
+    assert result.action_handler == "new_url"
+    await anc_app.occ_commands.unregister(result.name)
+    with pytest.raises(NextcloudExceptionNotFound):
+        await anc_app.occ_commands.unregister(result.name, not_fail=False)
+    await anc_app.occ_commands.unregister(result.name)
+    await anc_app.occ_commands.unregister(result2.name, not_fail=False)
+    assert await anc_app.occ_commands.get_entry(result2.name) is None
+    assert str(result).find("name=") != -1