feat: Add enum and default value support in task processing (#284)

All related PRs:
#284
nextcloud/app_api#359
nextcloud/translate2#11

---------

Signed-off-by: Anupam Kumar <[email protected]>
Signed-off-by: Alexander Piskun <[email protected]>
Co-authored-by: Alexander Piskun <[email protected]>

Author: kyteinsky

CHANGELOG.md

@@ -2,7 +2,11 @@
 
 All notable changes to this project will be documented in this file.
 
-## [0.16.0 - 2024-08-xx]
+## [0.16.0 - 2024-08-12]
+
+### Changed
+
+- NextcloudApp: rework of TaskProcessing provider API. #284
 
 ### Fixed
 

docs/reference/ExApp.rst

@@ -87,6 +87,21 @@ 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.providers.task_processing.ShapeType
+    :members:
+
+.. autoclass:: nc_py_api.ex_app.providers.task_processing.ShapeEnumValue
+    :members:
+
+.. autoclass:: nc_py_api.ex_app.providers.task_processing.ShapeDescriptor
+    :members:
+
+.. autoclass:: nc_py_api.ex_app.providers.task_processing.TaskType
+    :members:
+
+.. autoclass:: nc_py_api.ex_app.providers.task_processing.TaskProcessingProvider
+    :members:
+
 .. autoclass:: nc_py_api.ex_app.providers.task_processing._TaskProcessingProviderAPI
     :members:
 

nc_py_api/ex_app/providers/task_processing.py

@@ -3,35 +3,101 @@
 import contextlib
 import dataclasses
 import typing
+from enum import IntEnum
+
+from pydantic import RootModel
+from pydantic.dataclasses import dataclass
 
 from ..._exceptions import NextcloudException, NextcloudExceptionNotFound
-from ..._misc import clear_from_params_empty, require_capabilities
+from ..._misc import require_capabilities
 from ..._session import AsyncNcSessionApp, NcSessionApp
 
 _EP_SUFFIX: str = "ai_provider/task_processing"
 
 
-@dataclasses.dataclass
-class TaskProcessingProvider:
-    """TaskProcessing provider description."""
+class ShapeType(IntEnum):
+    """Enum for shape types."""
+
+    NUMBER = 0
+    TEXT = 1
+    IMAGE = 2
+    AUDIO = 3
+    VIDEO = 4
+    FILE = 5
+    ENUM = 6
+    LIST_OF_NUMBERS = 10
+    LIST_OF_TEXTS = 11
+    LIST_OF_IMAGES = 12
+    LIST_OF_AUDIOS = 13
+    LIST_OF_VIDEOS = 14
+    LIST_OF_FILES = 15
+
+
+@dataclass
+class ShapeEnumValue:
+    """Data object for input output shape enum slot value."""
+
+    name: str
+    """Name of the enum slot value which will be displayed in the UI"""
+    value: str
+    """Value of the enum slot value"""
+
 
-    def __init__(self, raw_data: dict):
-        self._raw_data = raw_data
+@dataclass
+class ShapeDescriptor:
+    """Data object for input output shape entries."""
 
-    @property
-    def name(self) -> str:
-        """Unique ID for the provider."""
-        return self._raw_data["name"]
+    name: str
+    """Name of the shape entry"""
+    description: str
+    """Description of the shape entry"""
+    shape_type: ShapeType
+    """Type of the shape entry"""
 
-    @property
-    def display_name(self) -> str:
-        """Providers display name."""
-        return self._raw_data["display_name"]
 
-    @property
-    def task_type(self) -> str:
-        """The TaskType provided by this provider."""
-        return self._raw_data["task_type"]
+@dataclass
+class TaskType:
+    """TaskType description for the provider."""
+
+    id: str
+    """The unique ID for the task type."""
+    name: str
+    """The localized name of the task type."""
+    description: str
+    """The localized description of the task type."""
+    input_shape: list[ShapeDescriptor]
+    """The input shape of the task."""
+    output_shape: list[ShapeDescriptor]
+    """The output shape of the task."""
+
+
+@dataclass
+class TaskProcessingProvider:
+
+    id: str
+    """Unique ID for the provider."""
+    name: str
+    """The localized name of this provider"""
+    task_type: str
+    """The TaskType provided by this provider."""
+    expected_runtime: int = dataclasses.field(default=0)
+    """Expected runtime of the task in seconds."""
+    optional_input_shape: list[ShapeDescriptor] = dataclasses.field(default_factory=list)
+    """Optional input shape of the task."""
+    optional_output_shape: list[ShapeDescriptor] = dataclasses.field(default_factory=list)
+    """Optional output shape of the task."""
+    input_shape_enum_values: dict[str, list[ShapeEnumValue]] = dataclasses.field(default_factory=dict)
+    """The option dict for each input shape ENUM slot."""
+    input_shape_defaults: dict[str, str | int | float] = dataclasses.field(default_factory=dict)
+    """The default values for input shape slots."""
+    optional_input_shape_enum_values: dict[str, list[ShapeEnumValue]] = dataclasses.field(default_factory=dict)
+    """The option list for each optional input shape ENUM slot."""
+    optional_input_shape_defaults: dict[str, str | int | float] = dataclasses.field(default_factory=dict)
+    """The default values for optional input shape slots."""
+    output_shape_enum_values: dict[str, list[ShapeEnumValue]] = dataclasses.field(default_factory=dict)
+    """The option list for each output shape ENUM slot."""
+    optional_output_shape_enum_values: dict[str, list[ShapeEnumValue]] = dataclasses.field(default_factory=dict)
+    """The option list for each optional output shape ENUM slot."""
 
     def __repr__(self):
         return f"<{self.__class__.__name__} name={self.name}, type={self.task_type}>"
@@ -44,17 +110,16 @@ def __init__(self, session: NcSessionApp):
         self._session = session
 
     def register(
-        self, name: str, display_name: str, task_type: str, custom_task_type: dict[str, typing.Any] | None = None
+        self,
+        provider: TaskProcessingProvider,
+        custom_task_type: TaskType | 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,
+            "provider": RootModel(provider).model_dump(),
+            **({"customTaskType": RootModel(custom_task_type).model_dump()} if custom_task_type else {}),
         }
-        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:
@@ -123,17 +188,16 @@ def __init__(self, session: AsyncNcSessionApp):
         self._session = session
 
     async def register(
-        self, name: str, display_name: str, task_type: str, custom_task_type: dict[str, typing.Any] | None = None
+        self,
+        provider: TaskProcessingProvider,
+        custom_task_type: TaskType | 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,
+            "provider": RootModel(provider).model_dump(),
+            **({"customTaskType": RootModel(custom_task_type).model_dump()} if custom_task_type else {}),
         }
-        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:

pyproject.toml

@@ -129,6 +129,7 @@ lint.select = [
   "W",
 ]
 lint.extend-ignore = [
+  "D101",
   "D105",
   "D107",
   "D203",
@@ -173,7 +174,6 @@ master.py-version = "3.10"
 master.extension-pkg-allow-list = [
   "pydantic",
 ]
-design.max-attributes = 8
 design.max-locals = 20
 design.max-branches = 16
 design.max-returns = 8
@@ -206,6 +206,7 @@ messages_control.disable = [
   "line-too-long",
   "too-few-public-methods",
   "too-many-public-methods",
+  "too-many-instance-attributes",
 ]
 
 [tool.pytest.ini_options]