diff --git a/src/middlewared/middlewared/api/base/model.py b/src/middlewared/middlewared/api/base/model.py
index 81067b3ad049a..327d238bc3d6c 100644
--- a/src/middlewared/middlewared/api/base/model.py
+++ b/src/middlewared/middlewared/api/base/model.py
@@ -9,8 +9,7 @@
 from pydantic.main import IncEx
 from typing_extensions import Annotated
 
-from middlewared.api.base.types.base import SECRET_VALUE
-from middlewared.api.base.types.base.string import LongStringWrapper
+from middlewared.api.base.types.string import SECRET_VALUE, LongStringWrapper
 from middlewared.utils.lang import undefined
 
 
diff --git a/src/middlewared/middlewared/api/base/types/__init__.py b/src/middlewared/middlewared/api/base/types/__init__.py
index 0cd96bab724a9..a7d4f619d98fe 100644
--- a/src/middlewared/middlewared/api/base/types/__init__.py
+++ b/src/middlewared/middlewared/api/base/types/__init__.py
@@ -1,5 +1,5 @@
-from .base import *  # noqa
 from .fc import *  # noqa
+from .filesystem import *  # noqa
 from .iscsi import *  # noqa
+from .string import *  # noqa
 from .user import *  # noqa
-from .filesystem import *  # noqa
diff --git a/src/middlewared/middlewared/api/base/types/base/__init__.py b/src/middlewared/middlewared/api/base/types/base/__init__.py
deleted file mode 100644
index 4524f9c9a187c..0000000000000
--- a/src/middlewared/middlewared/api/base/types/base/__init__.py
+++ /dev/null
@@ -1 +0,0 @@
-from .string import *  # noqa
diff --git a/src/middlewared/middlewared/api/base/types/base/string.py b/src/middlewared/middlewared/api/base/types/string.py
similarity index 81%
rename from src/middlewared/middlewared/api/base/types/base/string.py
rename to src/middlewared/middlewared/api/base/types/string.py
index e827b11c0ebea..41ea65af44b20 100644
--- a/src/middlewared/middlewared/api/base/types/base/string.py
+++ b/src/middlewared/middlewared/api/base/types/string.py
@@ -4,12 +4,15 @@
 from pydantic_core import CoreSchema, core_schema, PydanticKnownError
 from typing_extensions import Annotated
 
+from middlewared.api.base.validators import time_validator
 from middlewared.utils.netbios import validate_netbios_name, validate_netbios_domain
-from middlewared.validators import Time
+from zettarepl.snapshot.name import validate_snapshot_naming_schema
 
-__all__ = ["HttpUrl", "LongString", "NonEmptyString", "LongNonEmptyString", "SECRET_VALUE", "TimeString", "NetbiosDomain", "NetbiosName"]
 
-HttpUrl = Annotated[_HttpUrl, AfterValidator(str)]
+__all__ = [
+    "HttpUrl", "LongString", "NonEmptyString", "LongNonEmptyString", "SECRET_VALUE", "TimeString", "NetbiosDomain",
+    "NetbiosName", "SnapshotNameSchema"
+]
 
 
 class LongStringWrapper:
@@ -47,17 +50,18 @@ def __get_pydantic_core_schema__(
         )
 
 
+HttpUrl = Annotated[_HttpUrl, AfterValidator(str)]
 # By default, our strings are no more than 1024 characters long. This string is 2**31-1 characters long (SQLite limit).
 LongString = Annotated[
     LongStringWrapper,
     BeforeValidator(LongStringWrapper),
     PlainSerializer(lambda x: x.value if isinstance(x, LongStringWrapper) else x),
 ]
-
 NonEmptyString = Annotated[str, Field(min_length=1)]
 LongNonEmptyString = Annotated[LongString, Field(min_length=1)]
-TimeString = Annotated[str, AfterValidator(Time())]
+TimeString = Annotated[str, AfterValidator(time_validator)]
 NetbiosDomain = Annotated[str, AfterValidator(validate_netbios_domain)]
 NetbiosName = Annotated[str, AfterValidator(validate_netbios_name)]
+SnapshotNameSchema = Annotated[str, AfterValidator(lambda val: validate_snapshot_naming_schema(val) or val)]
 
 SECRET_VALUE = "********"
diff --git a/src/middlewared/middlewared/api/base/validators.py b/src/middlewared/middlewared/api/base/validators.py
new file mode 100644
index 0000000000000..efe4df49e8f89
--- /dev/null
+++ b/src/middlewared/middlewared/api/base/validators.py
@@ -0,0 +1,23 @@
+from datetime import time
+import re
+
+
+def match_validator(pattern: re.Pattern, explanation: str | None = None):
+    def validator(value: str):
+        assert (value is None or pattern.match(value)), (explanation or f"Value does not match {pattern!r} pattern")
+        return value
+
+    return validator
+
+
+def time_validator(value: str):
+    try:
+        hours, minutes = value.split(':')
+    except ValueError:
+        raise ValueError('Time should be in 24 hour format like "18:00"')
+    else:
+        try:
+            time(int(hours), int(minutes))
+        except TypeError:
+            raise ValueError('Time should be in 24 hour format like "18:00"')
+    return value
diff --git a/src/middlewared/middlewared/api/base/validators/__init__.py b/src/middlewared/middlewared/api/base/validators/__init__.py
deleted file mode 100644
index 4524f9c9a187c..0000000000000
--- a/src/middlewared/middlewared/api/base/validators/__init__.py
+++ /dev/null
@@ -1 +0,0 @@
-from .string import *  # noqa
diff --git a/src/middlewared/middlewared/api/base/validators/string.py b/src/middlewared/middlewared/api/base/validators/string.py
deleted file mode 100644
index 62d2e4bbc939d..0000000000000
--- a/src/middlewared/middlewared/api/base/validators/string.py
+++ /dev/null
@@ -1,10 +0,0 @@
-import re
-
-__all__ = ["match_validator"]
-
-
-def match_validator(pattern: re.Pattern, explanation: str | None = None):
-    def validator(value: str):
-        assert (value is None or pattern.match(value)), (explanation or f"Value does not match {pattern!r} pattern")
-
-    return validator
diff --git a/src/middlewared/middlewared/api/v25_04_0/__init__.py b/src/middlewared/middlewared/api/v25_04_0/__init__.py
index fd361cd3a2e5e..ae75272153db0 100644
--- a/src/middlewared/middlewared/api/v25_04_0/__init__.py
+++ b/src/middlewared/middlewared/api/v25_04_0/__init__.py
@@ -25,9 +25,10 @@
 from .iscsi_extent import *  # noqa
 from .keychain import *  # noqa
 from .netdata import *  # noqa
-from .pool_scrub import *  # noqa
 from .pool import *  # noqa
 from .pool_resilver import *  # noqa
+from .pool_scrub import *  # noqa
+from .pool_snapshottask import *  # noqa
 from .privilege import *  # noqa
 from .reporting import *  # noqa
 from .reporting_exporters import *  # noqa
diff --git a/src/middlewared/middlewared/api/v25_04_0/pool_snapshottask.py b/src/middlewared/middlewared/api/v25_04_0/pool_snapshottask.py
new file mode 100644
index 0000000000000..e8cc56d67944f
--- /dev/null
+++ b/src/middlewared/middlewared/api/v25_04_0/pool_snapshottask.py
@@ -0,0 +1,120 @@
+from typing import Any, Literal
+
+from pydantic import Field
+
+from middlewared.api.base import BaseModel, ForUpdateMetaclass, TimeString, SnapshotNameSchema
+from .common import CronModel
+
+
+__all__ = [
+    "PoolSnapshotTaskEntry", "PoolSnapshotTaskCreateArgs", "PoolSnapshotTaskCreateResult",
+    "PoolSnapshotTaskUpdateArgs", "PoolSnapshotTaskUpdateResult", "PoolSnapshotTaskDeleteArgs",
+    "PoolSnapshotTaskDeleteResult", "PoolSnapshotTaskMaxCountArgs", "PoolSnapshotTaskMaxCountResult",
+    "PoolSnapshotTaskMaxTotalCountArgs", "PoolSnapshotTaskMaxTotalCountResult", "PoolSnapshotTaskRunArgs",
+    "PoolSnapshotTaskRunResult", "PoolSnapshotTaskUpdateWillChangeRetentionForArgs",
+    "PoolSnapshotTaskUpdateWillChangeRetentionForResult", "PoolSnapshotTaskDeleteWillChangeRetentionForArgs",
+    "PoolSnapshotTaskDeleteWillChangeRetentionForResult"
+]
+
+
+class PoolSnapshotTaskCron(CronModel):
+    minute: str = "00"
+    begin: TimeString = "00:00"
+    end: TimeString = "23:59"
+
+
+class PoolSnapshotTaskCreate(BaseModel):
+    dataset: str
+    recursive: bool = False
+    lifetime_value: int = 2
+    lifetime_unit: Literal["HOUR", "DAY", "WEEK", "MONTH", "YEAR"] = "WEEK"
+    enabled: bool = True
+    exclude: list[str] = []
+    naming_schema: SnapshotNameSchema = "auto-%Y-%m-%d_%H-%M"
+    allow_empty: bool = True
+    schedule: PoolSnapshotTaskCron = Field(default_factory=PoolSnapshotTaskCron)
+
+
+class PoolSnapshotTaskUpdate(PoolSnapshotTaskCreate, metaclass=ForUpdateMetaclass):
+    fixate_removal_date: bool
+
+
+class PoolSnapshotTaskUpdateWillChangeRetentionFor(PoolSnapshotTaskCreate, metaclass=ForUpdateMetaclass):
+    pass
+
+
+class PoolSnapshotTaskDeleteOptions(BaseModel):
+    fixate_removal_date: bool = False
+
+
+class PoolSnapshotTaskEntry(PoolSnapshotTaskCreate):
+    id: int
+    vmware_sync: bool
+    state: Any
+
+
+class PoolSnapshotTaskCreateArgs(BaseModel):
+    data: PoolSnapshotTaskCreate
+
+
+class PoolSnapshotTaskCreateResult(BaseModel):
+    result: PoolSnapshotTaskEntry
+
+
+class PoolSnapshotTaskUpdateArgs(BaseModel):
+    id: int
+    data: PoolSnapshotTaskUpdate
+
+
+class PoolSnapshotTaskUpdateResult(BaseModel):
+    result: PoolSnapshotTaskEntry
+
+
+class PoolSnapshotTaskDeleteArgs(BaseModel):
+    id: int
+    options: PoolSnapshotTaskDeleteOptions = Field(default_factory=PoolSnapshotTaskDeleteOptions)
+
+
+class PoolSnapshotTaskDeleteResult(BaseModel):
+    result: Literal[True]
+
+
+class PoolSnapshotTaskMaxCountArgs(BaseModel):
+    pass
+
+
+class PoolSnapshotTaskMaxCountResult(BaseModel):
+    result: int
+
+
+class PoolSnapshotTaskMaxTotalCountArgs(BaseModel):
+    pass
+
+
+class PoolSnapshotTaskMaxTotalCountResult(BaseModel):
+    result: int
+
+
+class PoolSnapshotTaskRunArgs(BaseModel):
+    id: int
+
+
+class PoolSnapshotTaskRunResult(BaseModel):
+    result: None
+
+
+class PoolSnapshotTaskUpdateWillChangeRetentionForArgs(BaseModel):
+    id: int
+    data: PoolSnapshotTaskUpdateWillChangeRetentionFor
+
+
+class PoolSnapshotTaskUpdateWillChangeRetentionForResult(BaseModel):
+    result: dict[str, list[str]]
+
+
+class PoolSnapshotTaskDeleteWillChangeRetentionForArgs(BaseModel):
+    id: int
+
+
+class PoolSnapshotTaskDeleteWillChangeRetentionForResult(BaseModel):
+    result: dict[str, list[str]]
diff --git a/src/middlewared/middlewared/plugins/snapshot.py b/src/middlewared/middlewared/plugins/snapshot.py
index 26807f5f3a1df..7a05bd6234b62 100644
--- a/src/middlewared/middlewared/plugins/snapshot.py
+++ b/src/middlewared/middlewared/plugins/snapshot.py
@@ -1,13 +1,18 @@
-from datetime import datetime, time, timedelta
+from datetime import time
 import os
 
+from middlewared.api import api_method
+from middlewared.api.current import (
+    PoolSnapshotTaskEntry, PoolSnapshotTaskCreateArgs, PoolSnapshotTaskCreateResult, PoolSnapshotTaskUpdateArgs,
+    PoolSnapshotTaskUpdateResult, PoolSnapshotTaskDeleteArgs, PoolSnapshotTaskDeleteResult,
+    PoolSnapshotTaskMaxCountArgs, PoolSnapshotTaskMaxCountResult, PoolSnapshotTaskMaxTotalCountArgs,
+    PoolSnapshotTaskMaxTotalCountResult, PoolSnapshotTaskRunArgs, PoolSnapshotTaskRunResult
+)
 from middlewared.common.attachment import FSAttachmentDelegate
-from middlewared.schema import accepts, returns, Bool, Cron, Dataset, Dict, Int, List, Patch, Str
+from middlewared.schema import Cron
 from middlewared.service import CallError, CRUDService, item_method, private, ValidationErrors
 import middlewared.sqlalchemy as sa
-from middlewared.utils.cron import croniter_for_schedule
 from middlewared.utils.path import is_child
-from middlewared.validators import ReplicationSnapshotNamingSchema
 
 
 class PeriodicSnapshotTaskModel(sa.Model):
@@ -40,6 +45,7 @@ class Config:
         datastore_extend_context = 'pool.snapshottask.extend_context'
         namespace = 'pool.snapshottask'
         cli_namespace = 'task.snapshot'
+        entry = PoolSnapshotTaskEntry
 
     @private
     async def extend_context(self, rows, extra):
@@ -69,29 +75,9 @@ async def extend(self, data, context):
 
         return data
 
-    @accepts(
-        Dict(
-            'periodic_snapshot_create',
-            Dataset('dataset', required=True),
-            Bool('recursive', required=True),
-            List('exclude', items=[Dataset('item')]),
-            Int('lifetime_value', required=True),
-            Str('lifetime_unit', enum=['HOUR', 'DAY', 'WEEK', 'MONTH', 'YEAR'], required=True),
-            Str('naming_schema', required=True, validators=[ReplicationSnapshotNamingSchema()]),
-            Cron(
-                'schedule',
-                defaults={
-                    'minute': '00',
-                    'begin': '00:00',
-                    'end': '23:59',
-                },
-                required=True,
-                begin_end=True
-            ),
-            Bool('allow_empty', default=True),
-            Bool('enabled', default=True),
-            register=True
-        ),
+    @api_method(
+        PoolSnapshotTaskCreateArgs,
+        PoolSnapshotTaskCreateResult,
         audit='Snapshot task create:',
         audit_extended=lambda data: data['dataset']
     )
@@ -158,16 +144,11 @@ async def do_create(self, data):
 
         return await self.get_instance(data['id'])
 
-    @accepts(
-        Int('id', required=True),
-        Patch(
-            'periodic_snapshot_create',
-            'periodic_snapshot_update',
-            ('add', {'name': 'fixate_removal_date', 'type': 'bool'}),
-            ('attr', {'update': True})
-        ),
+    @api_method(
+        PoolSnapshotTaskUpdateArgs,
+        PoolSnapshotTaskUpdateResult,
         audit='Snapshot task update:',
-        audit_callback=True,
+        audit_callback=True
     )
     async def do_update(self, audit_callback, id_, data):
         """
@@ -255,14 +236,11 @@ async def do_update(self, audit_callback, id_, data):
 
         return await self.get_instance(id_)
 
-    @accepts(
-        Int('id'),
-        Dict(
-            'options',
-            Bool('fixate_removal_date', default=False),
-        ),
+    @api_method(
+        PoolSnapshotTaskDeleteArgs,
+        PoolSnapshotTaskDeleteResult,
         audit='Snapshot task delete:',
-        audit_callback=True,
+        audit_callback=True
     )
     async def do_delete(self, audit_callback, id_, options):
         """
@@ -318,8 +296,7 @@ async def do_delete(self, audit_callback, id_, options):
 
         return response
 
-    @accepts()
-    @returns(Int())
+    @api_method(PoolSnapshotTaskMaxCountArgs, PoolSnapshotTaskMaxCountResult)
     def max_count(self):
         """
         Returns a maximum amount of snapshots (per-dataset) the system can sustain.
@@ -329,8 +306,7 @@ def max_count(self):
         # with too many, then File Explorer will show no snapshots available.
         return 512
 
-    @accepts()
-    @returns(Int())
+    @api_method(PoolSnapshotTaskMaxTotalCountArgs, PoolSnapshotTaskMaxTotalCountResult)
     def max_total_count(self):
         """
         Returns a maximum amount of snapshots (total) the system can sustain.
@@ -341,7 +317,7 @@ def max_total_count(self):
         return 10000
 
     @item_method
-    @accepts(Int("id"))
+    @api_method(PoolSnapshotTaskRunArgs, PoolSnapshotTaskRunResult)
     async def run(self, id_):
         """
         Execute a Periodic Snapshot Task of `id`.
diff --git a/src/middlewared/middlewared/plugins/snapshot_/removal_date.py b/src/middlewared/middlewared/plugins/snapshot_/removal_date.py
index 6bd9b7e90233c..4b3b01ab3c42a 100644
--- a/src/middlewared/middlewared/plugins/snapshot_/removal_date.py
+++ b/src/middlewared/middlewared/plugins/snapshot_/removal_date.py
@@ -1,5 +1,3 @@
-import hashlib
-
 from middlewared.service import Service, job, private
 
 
diff --git a/src/middlewared/middlewared/plugins/snapshot_/task_retention.py b/src/middlewared/middlewared/plugins/snapshot_/task_retention.py
index 561645695b4cf..267d1e8408a9b 100644
--- a/src/middlewared/middlewared/plugins/snapshot_/task_retention.py
+++ b/src/middlewared/middlewared/plugins/snapshot_/task_retention.py
@@ -1,7 +1,11 @@
 from collections import defaultdict
 
-from middlewared.schema import Dict, Int, Patch, returns
-from middlewared.service import accepts, item_method, Service
+from middlewared.api import api_method
+from middlewared.api.current import (
+    PoolSnapshotTaskUpdateWillChangeRetentionForArgs, PoolSnapshotTaskUpdateWillChangeRetentionForResult,
+    PoolSnapshotTaskDeleteWillChangeRetentionForArgs, PoolSnapshotTaskDeleteWillChangeRetentionForResult
+)
+from middlewared.service import item_method, Service
 
 
 class PeriodicSnapshotTaskService(Service):
@@ -10,15 +14,7 @@ class Config:
         namespace = "pool.snapshottask"
 
     @item_method
-    @accepts(
-        Int("id"),
-        Patch(
-            "periodic_snapshot_create",
-            "periodic_snapshot_update_will_change_retention",
-            ("attr", {"update": True}),
-        ),
-    )
-    @returns(Dict("snapshots", additional_attrs=True))
+    @api_method(PoolSnapshotTaskUpdateWillChangeRetentionForArgs, PoolSnapshotTaskUpdateWillChangeRetentionForResult)
     async def update_will_change_retention_for(self, id_, data):
         """
         Returns a list of snapshots which will change the retention if periodic snapshot task `id` is updated
@@ -40,10 +36,7 @@ async def update_will_change_retention_for(self, id_, data):
         return result
 
     @item_method
-    @accepts(
-        Int("id"),
-    )
-    @returns(Dict("snapshots", additional_attrs=True))
+    @api_method(PoolSnapshotTaskDeleteWillChangeRetentionForArgs, PoolSnapshotTaskDeleteWillChangeRetentionForResult)
     async def delete_will_change_retention_for(self, id_):
         """
         Returns a list of snapshots which will change the retention if periodic snapshot task `id` is deleted.