diff --git a/src/middlewared/middlewared/plugins/keyvalue.py b/src/middlewared/middlewared/plugins/keyvalue.py
index dae5fe7e505f..637e4eea620d 100644
--- a/src/middlewared/middlewared/plugins/keyvalue.py
+++ b/src/middlewared/middlewared/plugins/keyvalue.py
@@ -1,6 +1,7 @@
+from typing import Any
+
 from truenas_api_client import json
 
-from middlewared.schema import Any, Str, accepts, Dict
 from middlewared.service import Service
 import middlewared.sqlalchemy as sa
 
@@ -18,16 +19,14 @@ class KeyValueService(Service):
     class Config:
         private = True
 
-    @accepts(Str('key'))
-    async def has_key(self, key):
+    async def has_key(self, key: str):
         try:
             await self.get(key)
             return True
         except KeyError:
             return False
 
-    @accepts(Str('key'), Any('default', null=True, default=None))
-    async def get(self, key, default):
+    async def get(self, key: str, default: Any | None = None):
         try:
             return json.loads(
                 (await self.middleware.call(
@@ -38,28 +37,21 @@ async def get(self, key, default):
 
             raise KeyError(key)
 
-    @accepts(
-        Str('key'),
-        Any('value'),
-        Dict('options', additional_attrs=True),
-    )
-    async def set(self, key, value, options):
+    async def set(self, key: str, value: Any, options: dict | None = None):
+        opts = options if options is not None else dict()
         try:
             row = await self.middleware.call("datastore.query", "system.keyvalue", [["key", "=", key]], {"get": True})
         except IndexError:
             await self.middleware.call(
-                "datastore.insert", "system.keyvalue", {"key": key, "value": json.dumps(value)}, options
+                "datastore.insert", "system.keyvalue", {"key": key, "value": json.dumps(value)}, opts
             )
         else:
             await self.middleware.call(
-                "datastore.update", "system.keyvalue", row["id"], {"value": json.dumps(value)}, options
+                "datastore.update", "system.keyvalue", row["id"], {"value": json.dumps(value)}, opts
             )
 
         return value
 
-    @accepts(
-        Str('key'),
-        Dict('options', additional_attrs=True),
-    )
-    async def delete(self, key, options):
-        await self.middleware.call("datastore.delete", "system.keyvalue", [["key", "=", key]], options)
+    async def delete(self, key: str, options: dict | None = None):
+        opts = options if options is not None else dict()
+        await self.middleware.call("datastore.delete", "system.keyvalue", [["key", "=", key]], opts)