added report_result method to TextProcessing providers API

nc_py_api/ex_app/providers/text_processing.py

@@ -1,8 +1,9 @@
 """Nextcloud API for declaring TextProcessing provider."""
 
+import contextlib
 import dataclasses
 
-from ..._exceptions import NextcloudExceptionNotFound
+from ..._exceptions import NextcloudException, NextcloudExceptionNotFound
 from ..._misc import require_capabilities
 from ..._session import AsyncNcSessionApp, NcSessionApp
 
@@ -76,6 +77,16 @@ def get_entry(self, name: str) -> TextProcessingProvider | None:
         except NextcloudExceptionNotFound:
             return None
 
+    def report_result(self, task_id: int, result: str = "", error: str = "") -> None:
+        """Report results of the text processing to Nextcloud."""
+        require_capabilities("app_api", self._session.capabilities)
+        with contextlib.suppress(NextcloudException):
+            self._session.ocs(
+                "PUT",
+                f"{self._session.ae_url}/{self._ep_suffix}",
+                params={"taskId": task_id, "result": result, "error": error},
+            )
+
 
 class _AsyncTextProcessingProviderAPI:
     """API for registering TextProcessing providers."""
@@ -114,3 +125,13 @@ async def get_entry(self, name: str) -> TextProcessingProvider | None:
             )
         except NextcloudExceptionNotFound:
             return None
+
+    async def report_result(self, task_id: int, result: str = "", error: str = "") -> None:
+        """Report results of the text processing to Nextcloud."""
+        require_capabilities("app_api", await self._session.capabilities)
+        with contextlib.suppress(NextcloudException):
+            await self._session.ocs(
+                "PUT",
+                f"{self._session.ae_url}/{self._ep_suffix}",
+                params={"taskId": task_id, "result": result, "error": error},
+            )

tests/actual_tests/text_processing_provider_test.py

@@ -56,3 +56,14 @@ async def test_text_processing_provider_async(anc_app):
     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
+
+
+@pytest.mark.require_nc(major=29)
+def test_text_processing_provider_fail_report(nc_app):
+    nc_app.providers.text_processing.report_result(999999)
+
+
+@pytest.mark.asyncio(scope="session")
+@pytest.mark.require_nc(major=29)
+async def test_text_processing_provider_fail_report_async(anc_app):
+    await anc_app.providers.text_processing.report_result(999999)