NAS-133206 / 25.10 / Convert boot.* to versioned API

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

@@ -9,6 +9,7 @@
 from .app_ix_volume import *  # noqa
 from .app_registry import *  # noqa
 from .auth import *  # noqa
+from .boot import *  # noqa
 from .boot_environments import *  # noqa
 from .catalog import *  # noqa
 from .cloud_backup import *  # noqa

src/middlewared/middlewared/api/v25_04_0/boot.py

@@ -0,0 +1,94 @@
+from typing import Literal
+
+from pydantic import Field, PositiveInt
+
+from middlewared.api.base import BaseModel, NotRequired
+
+
+__all__ = [
+    "BootGetDisksArgs", "BootGetDisksResult", "BootAttachArgs", "BootAttachResult", "BootDetachArgs",
+    "BootDetachResult", "BootReplaceArgs", "BootReplaceResult", "BootScrubArgs", "BootScrubResult",
+    "BootSetScrubIntervalArgs", "BootSetScrubIntervalResult", "BootUpdateInitramfsArgs", "BootUpdateInitramfsResult",
+    "BootFormatArgs", "BootFormatResult"
+]
+
+
+class BootAttachOptions(BaseModel):
+    expand: bool = False
+
+
+class BootFormatOptions(BaseModel):
+    size: int = NotRequired
+    legacy_schema: Literal["BIOS_ONLY", "EFI_ONLY", None] = None
+
+
+class BootUpdateInitramfsOptions(BaseModel):
+    database: str | None = None
+    force: bool = False
+
+
+class BootGetDisksArgs(BaseModel):
+    pass
+
+
+class BootGetDisksResult(BaseModel):
+    result: list[str]
+
+
+class BootAttachArgs(BaseModel):
+    dev: str
+    options: BootAttachOptions = Field(default_factory=BootAttachOptions)
+
+
+class BootAttachResult(BaseModel):
+    result: None
+
+
+class BootDetachArgs(BaseModel):
+    dev: str
+
+
+class BootDetachResult(BaseModel):
+    result: None
+
+
+class BootFormatArgs(BaseModel):
+    dev: str
+    options: BootFormatOptions = Field(default_factory=BootFormatOptions)
+
+
+class BootFormatResult(BaseModel):
+    result: None
+
+
+class BootReplaceArgs(BaseModel):
+    label: str
+    dev: str
+
+
+class BootReplaceResult(BaseModel):
+    result: None
+
+
+class BootScrubArgs(BaseModel):
+    pass
+
+
+class BootScrubResult(BaseModel):
+    result: None
+
+
+class BootSetScrubIntervalArgs(BaseModel):
+    interval: PositiveInt
+
+
+class BootSetScrubIntervalResult(BaseModel):
+    result: PositiveInt
+
+
+class BootUpdateInitramfsArgs(BaseModel):
+    options: BootUpdateInitramfsOptions = Field(default_factory=BootUpdateInitramfsOptions)
+
+
+class BootUpdateInitramfsResult(BaseModel):
+    result: bool

src/middlewared/middlewared/plugins/boot.py

@@ -1,12 +1,17 @@
 import asyncio
 import os
 
-from contextlib import asynccontextmanager
-from middlewared.schema import accepts, Bool, Dict, Int, List, Str, returns, Patch
+from middlewared.api import api_method
+from middlewared.api.current import (
+    BootGetDisksArgs, BootGetDisksResult, BootAttachArgs, BootAttachResult, BootDetachArgs,
+    BootDetachResult, BootReplaceArgs, BootReplaceResult, BootScrubArgs, BootScrubResult,
+    BootSetScrubIntervalArgs, BootSetScrubIntervalResult, BootUpdateInitramfsArgs, BootUpdateInitramfsResult
+)
+from middlewared.schema import accepts, returns, Patch
 from middlewared.service import CallError, Service, job, private
 from middlewared.utils import run
 from middlewared.utils.disks import valid_zfs_partition_uuids
-from middlewared.validators import Range
+
 
 BOOT_ATTACH_REPLACE_LOCK = 'boot_attach_replace'
 BOOT_POOL_NAME = BOOT_POOL_DISKS = None
@@ -62,8 +67,7 @@ async def clear_disks_cache(self):
         global BOOT_POOL_DISKS
         BOOT_POOL_DISKS = None
 
-    @accepts(roles=['READONLY_ADMIN'])
-    @returns(List('disks', items=[Str('disk')]))
+    @api_method(BootGetDisksArgs, BootGetDisksResult, roles=['DISK_READ'])
     async def get_disks(self):
         """
         Returns disks of the boot pool.
@@ -81,14 +85,7 @@ async def get_boot_type(self):
         # https://wiki.debian.org/UEFI
         return 'EFI' if os.path.exists('/sys/firmware/efi') else 'BIOS'
 
-    @accepts(
-        Str('dev'),
-        Dict(
-            'options',
-            Bool('expand', default=False),
-        ),
-    )
-    @returns()
+    @api_method(BootAttachArgs, BootAttachResult, roles=['DISK_WRITE'])
     @job(lock=BOOT_ATTACH_REPLACE_LOCK)
     async def attach(self, job, dev, options):
         """
@@ -138,8 +135,7 @@ async def attach(self, job, dev, options):
         await self.middleware.call('zfs.pool.online', BOOT_POOL_NAME, zfs_dev_part['name'], True)
         await self.update_initramfs()
 
-    @accepts(Str('dev'))
-    @returns()
+    @api_method(BootDetachArgs, BootDetachResult, roles=['DISK_WRITE'])
     async def detach(self, dev):
         """
         Detach given `dev` from boot pool.
@@ -148,8 +144,7 @@ async def detach(self, dev):
         await self.middleware.call('zfs.pool.detach', BOOT_POOL_NAME, dev, {'clear_label': True})
         await self.update_initramfs()
 
-    @accepts(Str('label'), Str('dev'))
-    @returns()
+    @api_method(BootReplaceArgs, BootReplaceResult, roles=['DISK_WRITE'])
     @job(lock=BOOT_ATTACH_REPLACE_LOCK)
     async def replace(self, job, label, dev):
         """
@@ -187,8 +182,7 @@ async def replace(self, job, label, dev):
         await self.middleware.call('boot.install_loader', dev)
         await self.update_initramfs()
 
-    @accepts()
-    @returns()
+    @api_method(BootScrubArgs, BootScrubResult, roles=['BOOT_ENV_WRITE'])
     @job(lock='boot_scrub')
     async def scrub(self, job):
         """
@@ -197,10 +191,7 @@ async def scrub(self, job):
         subjob = await self.middleware.call('pool.scrub.scrub', BOOT_POOL_NAME)
         return await job.wrap(subjob)
 
-    @accepts(
-        Int('interval', validators=[Range(min_=1)])
-    )
-    @returns(Int('interval'))
+    @api_method(BootSetScrubIntervalArgs, BootSetScrubIntervalResult, roles=['BOOT_ENV_WRITE'])
     async def set_scrub_interval(self, interval):
         """
         Set Automatic Scrub Interval value in days.
@@ -213,12 +204,7 @@ async def set_scrub_interval(self, interval):
         )
         return interval
 
-    @accepts(Dict(
-        'options',
-        Str('database', default=None, null=True),
-        Bool('force', default=False),
-    ))
-    @private
+    @api_method(BootUpdateInitramfsArgs, BootUpdateInitramfsResult, private=True)
     async def update_initramfs(self, options):
         """
         Returns true if initramfs was updated and false otherwise.

src/middlewared/middlewared/plugins/boot_/format.py

@@ -1,19 +1,12 @@
-from middlewared.schema import accepts, Dict, Int, Str
-from middlewared.service import CallError, private, Service
+from middlewared.api import api_method
+from middlewared.api.current import BootFormatArgs, BootFormatResult
+from middlewared.service import CallError, Service
 from middlewared.utils import run
 
 
 class BootService(Service):
 
-    @accepts(
-        Str('dev'),
-        Dict(
-            'options',
-            Int('size'),
-            Str('legacy_schema', enum=[None, 'BIOS_ONLY', 'EFI_ONLY'], null=True, default=None),
-        )
-    )
-    @private
+    @api_method(BootFormatArgs, BootFormatResult, private=True)
     async def format(self, dev, options):
         """
         Format a given disk `dev` using the appropriate partition layout