added TextProcessing Provider API

CHANGELOG.md

@@ -6,7 +6,8 @@ All notable changes to this project will be documented in this file.
 
 ### Added
 
-- API for registering Speech to Text provider(*avalaible from Nextcloud 29*). #196
+- NextcloudApp: API for registering Speech to Text providers(*avalaible from Nextcloud 29*). #196
+- NextcloudApp: API for registering Text Processing providers(*avalaible from Nextcloud 29*). #197
 
 ### Fixed
 

docs/reference/ExApp.rst

@@ -65,3 +65,9 @@ UI methods should be accessed with the help of :class:`~nc_py_api.nextcloud.Next
 
 .. autoclass:: nc_py_api.ex_app.providers.speech_to_text._SpeechToTextProviderAPI
  :members:
+
+.. autoclass:: nc_py_api.ex_app.providers.text_processing.TextProcessingProvider
+ :members:
+
+.. autoclass:: nc_py_api.ex_app.providers.text_processing._TextProcessingProviderAPI
+ :members:

nc_py_api/ex_app/providers/providers.py

@@ -2,23 +2,30 @@
 
 from ..._session import AsyncNcSessionApp, NcSessionApp
 from .speech_to_text import _AsyncSpeechToTextProviderAPI, _SpeechToTextProviderAPI
+from .text_processing import _AsyncTextProcessingProviderAPI, _TextProcessingProviderAPI
 
 
 class ProvidersApi:
  """Class that encapsulates all AI Providers functionality."""
 
  speech_to_text: _SpeechToTextProviderAPI
  """SpeechToText Provider API."""
+ text_processing: _TextProcessingProviderAPI
+ """TextProcessing Provider API."""
 
  def __init__(self, session: NcSessionApp):
  self.speech_to_text = _SpeechToTextProviderAPI(session)
+ self.text_processing = _TextProcessingProviderAPI(session)
 
 
 class AsyncProvidersApi:
  """Class that encapsulates all AI Providers functionality."""
 
  speech_to_text: _AsyncSpeechToTextProviderAPI
  """SpeechToText Provider API."""
+ text_processing: _AsyncTextProcessingProviderAPI
+ """TextProcessing Provider API."""
 
  def __init__(self, session: AsyncNcSessionApp):
  self.speech_to_text = _AsyncSpeechToTextProviderAPI(session)
+ self.text_processing = _AsyncTextProcessingProviderAPI(session)

nc_py_api/ex_app/providers/text_processing.py

@@ -0,0 +1,116 @@
+"""Nextcloud API for declaring TextProcessing provider."""
+
+import dataclasses
+
+from ..._exceptions import NextcloudExceptionNotFound
+from ..._misc import require_capabilities
+from ..._session import AsyncNcSessionApp, NcSessionApp
+
+
+@dataclasses.dataclass
+class TextProcessingProvider:
+ """TextProcessing provider description."""
+
+ def __init__(self, raw_data: dict):
+ self._raw_data = raw_data
+
+ @property
+ def name(self) -> str:
+ """Unique ID for the provider."""
+ return self._raw_data["name"]
+
+ @property
+ def display_name(self) -> str:
+ """Providers display name."""
+ return self._raw_data["display_name"]
+
+ @property
+ def action_handler(self) -> str:
+ """Relative ExApp url which will be called by Nextcloud."""
+ return self._raw_data["action_handler"]
+
+ @property
+ def task_type(self) -> str:
+ """The TaskType provided by this provider."""
+ return self._raw_data["task_type"]
+
+ def __repr__(self):
+ return f"<{self.__class__.__name__} name={self.name}, type={self.task_type}, handler={self.action_handler}>"
+
+
+class _TextProcessingProviderAPI:
+ """API for registering TextProcessing providers."""
+
+ _ep_suffix: str = "ai_provider/text_processing"
+
+ def __init__(self, session: NcSessionApp):
+ self._session = session
+
+ def register(self, name: str, display_name: str, callback_url: str, task_type: str) -> None:
+ """Registers or edit the TextProcessing provider."""
+ require_capabilities("app_api", self._session.capabilities)
+ params = {
+ "name": name,
+ "displayName": display_name,
+ "actionHandler": callback_url,
+ "taskType": task_type,
+ }
+ self._session.ocs("POST", f"{self._session.ae_url}/{self._ep_suffix}", json=params)
+
+ def unregister(self, name: str, not_fail=True) -> None:
+ """Removes TextProcessing provider."""
+ require_capabilities("app_api", self._session.capabilities)
+ try:
+ self._session.ocs("DELETE", f"{self._session.ae_url}/{self._ep_suffix}", params={"name": name})
+ except NextcloudExceptionNotFound as e:
+ if not not_fail:
+ raise e from None
+
+ def get_entry(self, name: str) -> TextProcessingProvider | None:
+ """Get information of the TextProcessing."""
+ require_capabilities("app_api", self._session.capabilities)
+ try:
+ return TextProcessingProvider(
+ self._session.ocs("GET", f"{self._session.ae_url}/{self._ep_suffix}", params={"name": name})
+ )
+ except NextcloudExceptionNotFound:
+ return None
+
+
+class _AsyncTextProcessingProviderAPI:
+ """API for registering TextProcessing providers."""
+
+ _ep_suffix: str = "ai_provider/text_processing"
+
+ def __init__(self, session: AsyncNcSessionApp):
+ self._session = session
+
+ async def register(self, name: str, display_name: str, callback_url: str, task_type: str) -> None:
+ """Registers or edit the TextProcessing provider."""
+ require_capabilities("app_api", await self._session.capabilities)
+ params = {
+ "name": name,
+ "displayName": display_name,
+ "actionHandler": callback_url,
+ "taskType": task_type,
+ }
+ await self._session.ocs("POST", f"{self._session.ae_url}/{self._ep_suffix}", json=params)
+
+ async def unregister(self, name: str, not_fail=True) -> None:
+ """Removes TextProcessing provider."""
+ require_capabilities("app_api", await self._session.capabilities)
+ try:
+ await self._session.ocs("DELETE", f"{self._session.ae_url}/{self._ep_suffix}", params={"name": name})
+ except NextcloudExceptionNotFound as e:
+ if not not_fail:
+ raise e from None
+
+ async def get_entry(self, name: str) -> TextProcessingProvider | None:
+ """Get information of the TextProcessing."""
+ require_capabilities("app_api", await self._session.capabilities)
+ try:
+ return TextProcessingProvider(
+ await self._session.ocs("GET", f"{self._session.ae_url}/{self._ep_suffix}", params={"name": name})
+ )
+ except NextcloudExceptionNotFound:
+ return None

tests/actual_tests/text_processing_provider_test.py

@@ -0,0 +1,58 @@
+import pytest
+
+from nc_py_api import NextcloudExceptionNotFound
+
+
+@pytest.mark.require_nc(major=29)
+def test_text_processing_provider(nc_app):
+ nc_app.providers.text_processing.register("test_id", "Test #1 Prov", "/some_url", "free_prompt")
+ result = nc_app.providers.text_processing.get_entry("test_id")
+ assert result.name == "test_id"
+ assert result.display_name == "Test #1 Prov"
+ assert result.action_handler == "some_url"
+ nc_app.providers.text_processing.register("test_id2", "Test #2 Prov", "some_url2", "free_prompt")
+ result2 = nc_app.providers.text_processing.get_entry("test_id2")
+ assert result2.name == "test_id2"
+ assert result2.display_name == "Test #2 Prov"
+ assert result2.action_handler == "some_url2"
+ nc_app.providers.text_processing.register("test_id", "Renamed", "/new_url", "free_prompt")
+ result = nc_app.providers.text_processing.get_entry("test_id")
+ assert result.name == "test_id"
+ assert result.display_name == "Renamed"
+ assert result.action_handler == "new_url"
+ assert result.task_type == "free_prompt"
+ nc_app.providers.text_processing.unregister(result.name)
+ nc_app.providers.text_processing.unregister(result.name)
+ with pytest.raises(NextcloudExceptionNotFound):
+ nc_app.providers.text_processing.unregister(result.name, not_fail=False)
+ nc_app.providers.text_processing.unregister(result2.name, not_fail=False)
+ assert nc_app.providers.text_processing.get_entry(result2.name) is None
+ assert str(result).find("type=free_prompt") != -1
+
+
+@pytest.mark.asyncio(scope="session")
+@pytest.mark.require_nc(major=29)
+async def test_text_processing_provider_async(anc_app):
+ await anc_app.providers.text_processing.register("test_id", "Test #1 Prov", "/some_url", "free_prompt")
+ result = await anc_app.providers.text_processing.get_entry("test_id")
+ assert result.name == "test_id"
+ assert result.display_name == "Test #1 Prov"
+ assert result.action_handler == "some_url"
+ await anc_app.providers.text_processing.register("test_id2", "Test #2 Prov", "some_url2", "free_prompt")
+ result2 = await anc_app.providers.text_processing.get_entry("test_id2")
+ assert result2.name == "test_id2"
+ assert result2.display_name == "Test #2 Prov"
+ assert result2.action_handler == "some_url2"
+ await anc_app.providers.text_processing.register("test_id", "Renamed", "/new_url", "free_prompt")
+ result = await anc_app.providers.text_processing.get_entry("test_id")
+ assert result.name == "test_id"
+ assert result.display_name == "Renamed"
+ assert result.action_handler == "new_url"
+ assert result.task_type == "free_prompt"
+ await anc_app.providers.text_processing.unregister(result.name)
+ await anc_app.providers.text_processing.unregister(result.name)
+ with pytest.raises(NextcloudExceptionNotFound):
+ await anc_app.providers.text_processing.unregister(result.name, not_fail=False)
+ await anc_app.providers.text_processing.unregister(result2.name, not_fail=False)
+ assert await anc_app.providers.text_processing.get_entry(result2.name) is None
+ assert str(result).find("type=free_prompt") != -1