feat(TaskProcessing): Allow defining custom task types (#270)

Counter part for nextcloud/app_api#324

---------

Signed-off-by: provokateurin <[email protected]>

Author: provokateurin

.github/workflows/analysis-coverage.yml

@@ -847,7 +847,7 @@ jobs:
           PHP_CLI_SERVER_WORKERS=2 php -S localhost:8080 &
 
       - name: Enable Notes
-        if: ${{ !startsWith(matrix.nextcloud, 'master') }}
+        if: ${{ !startsWith(matrix.nextcloud, 'stable27') && !startsWith(matrix.nextcloud, 'master') }}
         run: ./occ app:enable notes
 
       - name: Enable Files Locking

nc_py_api/ex_app/providers/task_processing.py

@@ -5,7 +5,7 @@
 import typing
 
 from ..._exceptions import NextcloudException, NextcloudExceptionNotFound
-from ..._misc import require_capabilities
+from ..._misc import clear_from_params_empty, require_capabilities
 from ..._session import AsyncNcSessionApp, NcSessionApp
 
 _EP_SUFFIX: str = "ai_provider/task_processing"
@@ -43,14 +43,18 @@ class _TaskProcessingProviderAPI:
     def __init__(self, session: NcSessionApp):
         self._session = session
 
-    def register(self, name: str, display_name: str, task_type: str) -> None:
+    def register(
+        self, name: str, display_name: str, task_type: str, custom_task_type: dict[str, typing.Any] | None = None
+    ) -> None:
         """Registers or edit the TaskProcessing provider."""
         require_capabilities("app_api", self._session.capabilities)
         params = {
             "name": name,
             "displayName": display_name,
             "taskType": task_type,
+            "customTaskType": custom_task_type,
         }
+        clear_from_params_empty(["customTaskType"], params)
         self._session.ocs("POST", f"{self._session.ae_url}/{_EP_SUFFIX}", json=params)
 
     def unregister(self, name: str, not_fail=True) -> None:
@@ -118,14 +122,18 @@ class _AsyncTaskProcessingProviderAPI:
     def __init__(self, session: AsyncNcSessionApp):
         self._session = session
 
-    async def register(self, name: str, display_name: str, task_type: str) -> None:
+    async def register(
+        self, name: str, display_name: str, task_type: str, custom_task_type: dict[str, typing.Any] | None = None
+    ) -> None:
         """Registers or edit the TaskProcessing provider."""
         require_capabilities("app_api", await self._session.capabilities)
         params = {
             "name": name,
             "displayName": display_name,
             "taskType": task_type,
+            "customTaskType": custom_task_type,
         }
+        clear_from_params_empty(["customTaskType"], 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: