Convert UPS plugin to use pydantic models

src/middlewared/middlewared/api/v25_10_0/__init__.py

@@ -76,6 +76,7 @@
 from .tn_connect import *  # noqa
 from .truecommand import *  # noqa
 from .truenas import *  # noqa
+from .ups import *  # noqa
 from .user import *  # noqa
 from .virt_device import *  # noqa
 from .virt_global import *  # noqa

src/middlewared/middlewared/api/v25_10_0/ups.py

@@ -0,0 +1,65 @@
+from typing import Literal
+
+from pydantic import Field
+
+from middlewared.api.base import (
+    BaseModel, Excluded, excluded_field, ForUpdateMetaclass, NonEmptyString, LongString,
+    single_argument_args,
+)
+
+
+__all__ = [
+    'UPSEntry', 'UPSPortChoicesArgs', 'UPSPortChoicesResult', 'UPSDriverChoicesArgs',
+    'UPSDriverChoicesResult', 'UPSUpdateArgs', 'UPSUpdateResult',
+]
+
+
+class UPSEntry(BaseModel):
+    powerdown: bool
+    rmonitor: bool
+    id: int
+    nocommwarntime: int | None
+    remoteport: int = Field(ge=1, le=65535)
+    shutdowntimer: int
+    hostsync: int = Field(ge=0)
+    description: str
+    driver: str
+    extrausers: LongString
+    identifier: NonEmptyString
+    mode: Literal['MASTER', 'SLAVE']
+    monpwd: str
+    monuser: NonEmptyString
+    options: LongString
+    optionsupsd: LongString
+    port: str
+    remotehost: str
+    shutdown: Literal['LOWBATT', 'BATT']
+    shutdowncmd: str | None
+    complete_identifier: str
+
+
+@single_argument_args('ups_update')
+class UPSUpdateArgs(UPSEntry, metaclass=ForUpdateMetaclass):
+    id: Excluded = excluded_field()
+    complete_identifier: Excluded = excluded_field()
+    monpwd: NonEmptyString
+
+
+class UPSUpdateResult(BaseModel):
+    result: UPSEntry
+
+
+class UPSPortChoicesArgs(BaseModel):
+    pass
+
+
+class UPSPortChoicesResult(BaseModel):
+    result: list[str]
+
+
+class UPSDriverChoicesArgs(BaseModel):
+    pass
+
+
+class UPSDriverChoicesResult(BaseModel):
+    result: dict[str, str]

src/middlewared/middlewared/plugins/ups.py

@@ -5,12 +5,15 @@
 import os
 import re
 
-from middlewared.schema import accepts, Bool, Dict, Int, List, Patch, returns, Str
+from middlewared.api import api_method
+from middlewared.api.current import (
+    UPSEntry, UPSUpdateArgs, UPSUpdateResult, UPSPortChoicesArgs, UPSPortChoicesResult,
+    UPSDriverChoicesArgs, UPSDriverChoicesResult,
+)
 from middlewared.service import private, SystemServiceService, ValidationErrors
 import middlewared.sqlalchemy as sa
 from middlewared.utils import run
 from middlewared.utils.serial import serial_port_choices
-from middlewared.validators import Range, Port
 
 
 RE_DRIVER_CHOICE = re.compile(r'(\S+)\s+(\S+=\S+)?\s*(?:\((.+)\))?$')
@@ -51,30 +54,6 @@ def drivers_available():
 
 class UPSService(SystemServiceService):
 
-    ENTRY = Dict(
-        'ups_entry',
-        Bool('powerdown', required=True),
-        Bool('rmonitor', required=True),
-        Int('id', required=True),
-        Int('nocommwarntime', null=True, required=True),
-        Int('remoteport', validators=[Port()], required=True),
-        Int('shutdowntimer', required=True),
-        Int('hostsync', validators=[Range(min_=0)], required=True),
-        Str('description', required=True),
-        Str('driver', required=True),
-        Str('extrausers', max_length=None, required=True),
-        Str('identifier', empty=False, required=True),
-        Str('mode', enum=['MASTER', 'SLAVE'], required=True),
-        Str('monpwd', empty=False, required=True),
-        Str('monuser', empty=False, required=True),
-        Str('options', max_length=None, required=True),
-        Str('optionsupsd', max_length=None, required=True),
-        Str('port', required=True),
-        Str('remotehost', required=True),
-        Str('shutdown', enum=['LOWBATT', 'BATT'], required=True),
-        Str('shutdowncmd', null=True, required=True),
-        Str('complete_identifier', required=True),
-    )
     LOGGED_ERRORS = []
 
     class Config:
@@ -85,6 +64,7 @@ class Config:
         service_verb = 'restart'
         cli_namespace = 'service.ups'
         role_prefix = 'SYSTEM_GENERAL'
+        entry = UPSEntry
 
     @private
     async def ups_config_extend(self, data):
@@ -94,8 +74,7 @@ async def ups_config_extend(self, data):
         data['complete_identifier'] = f'{data["identifier"]}@{host}:{data["remoteport"]}'
         return data
 
-    @accepts()
-    @returns(List(items=[Str('port_choice')]))
+    @api_method(UPSPortChoicesArgs, UPSPortChoicesResult, roles=['SYSTEM_GENERAL_READ'])
     async def port_choices(self):
         adv_config = await self.middleware.call('system.advanced.config')
         ports = [
@@ -115,8 +94,7 @@ def normalize_driver_string(self, driver_str):
         driver = driver.replace(' ', '\n\t')  # "genericups upstype=16"
         return f'driver = {driver}'
 
-    @accepts()
-    @returns(Dict(additional_attrs=True, example={'blazer_ser$CPM-800': 'WinPower ups 2 CPM-800 (blazer_ser)'}))
+    @api_method(UPSDriverChoicesArgs, UPSDriverChoicesResult, roles=['SYSTEM_GENERAL_READ'])
     def driver_choices(self):
         """
         Returns choices of UPS drivers supported by the system.
@@ -221,15 +199,7 @@ async def validate_data(self, data, schema):
         verrors.check()
         return data
 
-    @accepts(
-        Patch(
-            'ups_entry', 'ups_update',
-            ('rm', {'name': 'id'}),
-            ('rm', {'name': 'complete_identifier'}),
-            ('edit', {'name': 'monpwd', 'method': lambda x: setattr(x, 'empty', False)}),
-            ('attr', {'update': True}),
-        ),
-    )
+    @api_method(UPSUpdateArgs, UPSUpdateResult)
     async def do_update(self, data):
         """
         Update UPS Service Configuration.
@@ -278,9 +248,6 @@ async def dismiss_alerts(self):
         await self.middleware.call('alert.oneshot_delete', alerts)
 
     @private
-    @accepts(
-        Str('notify_type')
-    )
     async def upssched_event(self, notify_type):
         config = await self.config()
         upsc_identifier = config['complete_identifier']