diff --git a/CHANGELOG.md b/CHANGELOG.md index 762ee534..3f018efa 100644 --- a/CHANGELOG.md +++ b/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 diff --git a/docs/reference/ExApp.rst b/docs/reference/ExApp.rst index 89757127..1f5a2168 100644 --- a/docs/reference/ExApp.rst +++ b/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: diff --git a/nc_py_api/ex_app/occ_commands.py b/nc_py_api/ex_app/occ_commands.py new file mode 100644 index 00000000..e9ec0d0b --- /dev/null +++ b/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.**.""" + + 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.**.""" + + 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 diff --git a/nc_py_api/nextcloud.py b/nc_py_api/nextcloud.py index 113dd6db..cae4a42e 100644 --- a/nc_py_api/nextcloud.py +++ b/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.""" diff --git a/pyproject.toml b/pyproject.toml index 9f70fd2f..fcd2f141 100644 --- a/pyproject.toml +++ b/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", diff --git a/tests/actual_tests/occ_commands_test.py b/tests/actual_tests/occ_commands_test.py new file mode 100644 index 00000000..75a65de8 --- /dev/null +++ b/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