Remove accepts style decorator from k8s to docker plugin (#15172)

src/middlewared/middlewared/api/v25_04_0/__init__.py

@@ -24,6 +24,7 @@
 from .iscsi_auth import *  # noqa
 from .iscsi_extent import *  # noqa
 from .keychain import *  # noqa
+from .k8s_to_docker import *  # noqa
 from .netdata import *  # noqa
 from .pool import *  # noqa
 from .pool_resilver import *  # noqa

src/middlewared/middlewared/api/v25_04_0/k8s_to_docker.py

@@ -0,0 +1,61 @@
+from pydantic import Field, RootModel
+
+from middlewared.api.base import BaseModel, single_argument_result
+
+
+__all__ = [
+    'K8sToDockerListBackupsArgs', 'K8sToDockerListBackupsResult', 'K8sToDockerMigrateArgs', 'K8sToDockerMigrateResult',
+]
+
+
+class K8sToDockerListBackupsArgs(BaseModel):
+    kubernetes_pool: str
+
+
+class ReleaseDetails(BaseModel):
+    error: str | None = None
+    helm_secret: dict = Field(default_factory=dict)
+    release_secrets: dict = Field(default_factory=dict)
+    train: str | None = None
+    app_name: str | None = None
+    app_version: str | None = None
+    release_name: str
+    migrate_file_path: str | None = None
+
+
+class BackupDetails(BaseModel):
+    name: str
+    releases: list[ReleaseDetails]
+    skipped_releases: list[ReleaseDetails]
+    snapshot_name: str
+    created_on: str
+    backup_path: str
+
+
+class Backups(RootModel[dict[str, BackupDetails]]):
+    pass
+
+
+@single_argument_result
+class K8sToDockerListBackupsResult(BaseModel):
+    error: str | None
+    backups: Backups
+
+
+class MigrateOptions(BaseModel):
+    backup_name: str | None = None
+
+
+class K8sToDockerMigrateArgs(BaseModel):
+    kubernetes_pool: str
+    options: MigrateOptions = MigrateOptions()
+
+
+class AppMigrationDetails(BaseModel):
+    name: str
+    successfully_migrated: bool
+    error: str | None
+
+
+class K8sToDockerMigrateResult(BaseModel):
+    result: list[AppMigrationDetails]

src/middlewared/middlewared/plugins/kubernetes_to_docker/list_k8s_backups.py

@@ -1,6 +1,7 @@
 import os
 
-from middlewared.schema import accepts, Dict, returns, Str
+from middlewared.api import api_method
+from middlewared.api.current import K8sToDockerListBackupsArgs, K8sToDockerListBackupsResult
 from middlewared.service import job, Service
 
 from .list_utils import get_backup_dir, get_default_release_details, K8s_BACKUP_NAME_PREFIX, release_details
@@ -13,12 +14,7 @@ class Config:
         namespace = 'k8s_to_docker'
         cli_namespace = 'k8s_to_docker'
 
-    @accepts(Str('kubernetes_pool'), roles=['DOCKER_READ'])
-    @returns(Dict(
-        'backups',
-        Str('error', null=True),
-        Dict('backups', additional_attrs=True),
-    ))
+    @api_method(K8sToDockerListBackupsArgs, K8sToDockerListBackupsResult, roles=['DOCKER_READ'])
     @job(lock=lambda args: f'k8s_to_docker_list_backups_{args[0]}')
     def list_backups(self, job, kubernetes_pool):
         """

src/middlewared/middlewared/plugins/kubernetes_to_docker/migrate.py

@@ -2,9 +2,10 @@
 import os.path
 import shutil
 
+from middlewared.api import api_method
+from middlewared.api.current import K8sToDockerMigrateArgs, K8sToDockerMigrateResult
 from middlewared.plugins.apps.ix_apps.path import get_app_parent_volume_ds_name, get_installed_app_path
 from middlewared.plugins.docker.state_utils import DatasetDefaults
-from middlewared.schema import accepts, Bool, Dict, List, returns, Str
 from middlewared.service import CallError, job, Service
 
 from .migrate_config_utils import migrate_chart_release_config
@@ -20,23 +21,7 @@ class Config:
         namespace = 'k8s_to_docker'
         cli_namespace = 'k8s_to_docker'
 
-    @accepts(
-        Str('kubernetes_pool'),
-        Dict(
-            'options',
-            Str('backup_name', null=True, default=None),
-        ),
-        roles=['DOCKER_WRITE']
-    )
-    @returns(List(
-        'app_migration_details',
-        items=[Dict(
-            'app_migration_detail',
-            Str('name'),
-            Bool('successfully_migrated'),
-            Str('error', null=True),
-        )]
-    ))
+    @api_method(K8sToDockerMigrateArgs, K8sToDockerMigrateResult, roles=['DOCKER_WRITE'])
     @job(lock='k8s_to_docker_migrate')
     def migrate(self, job, kubernetes_pool, options):
         """