extract interval validation to separate module

Author: so-saf

taskiq/scheduler/scheduled_task/v1.py

@@ -4,6 +4,8 @@
 
 from pydantic import BaseModel, Field, root_validator
 
+from taskiq.scheduler.scheduled_task.validators import validate_interval_value
+
 
 class ScheduledTask(BaseModel):
     """Abstraction over task schedule."""
@@ -34,26 +36,5 @@ def __check(cls, values: Dict[str, Any]) -> Dict[str, Any]:
         if all(values.get(key) is None for key in ("cron", "interval", "time")):
             raise ValueError("Either cron, interval, or datetime must be present.")
 
-        # Validate interval constraints
-        if (interval := values.get("interval")) is not None:
-            if isinstance(interval, int):
-                if interval < 1:
-                    raise ValueError(
-                        f"Interval must be at least 1 second, got {interval} seconds",
-                    )
-            else:
-                # For timedelta, check that it's at least 1 second
-                # and has no fractional seconds
-                total_seconds = interval.total_seconds()
-                if total_seconds != int(total_seconds):
-                    raise ValueError(
-                        f"Fractional intervals are not supported, "
-                        f"got {total_seconds} seconds",
-                    )
-                if total_seconds < 1:
-                    raise ValueError(
-                        f"Interval must be at least 1 second, "
-                        f"got {total_seconds} seconds",
-                    )
-
+        validate_interval_value(values.get("interval"))
         return values

taskiq/scheduler/scheduled_task/v2.py

@@ -5,6 +5,8 @@
 
 from pydantic import BaseModel, Field, model_validator
 
+from taskiq.scheduler.scheduled_task.validators import validate_interval_value
+
 if sys.version_info >= (3, 11):
     from typing import Self
 else:
@@ -39,27 +41,5 @@ def __check(self) -> Self:
         if self.cron is None and self.time is None and self.interval is None:
             raise ValueError("Either cron, interval, or datetime must be present.")
 
-        # Validate interval constraints
-        if self.interval is not None:
-            if isinstance(self.interval, int):
-                if self.interval < 1:
-                    raise ValueError(
-                        f"Interval must be at least 1 second, "
-                        f"got {self.interval} seconds",
-                    )
-            else:
-                # For timedelta, check that it's at least 1 second
-                # and has no fractional seconds
-                total_seconds = self.interval.total_seconds()
-                if total_seconds != int(total_seconds):
-                    raise ValueError(
-                        f"Fractional intervals are not supported, "
-                        f"got {total_seconds} seconds",
-                    )
-                if total_seconds < 1:
-                    raise ValueError(
-                        f"Interval must be at least 1 second, "
-                        f"got {total_seconds} seconds",
-                    )
-
+        validate_interval_value(self.interval)
         return self

taskiq/scheduler/scheduled_task/validators.py

@@ -0,0 +1,32 @@
+from datetime import timedelta
+from typing import Union
+
+
+def validate_interval_value(
+    value: Union[int, timedelta, None],
+) -> None:
+    """Validate that the given interval value meets required constraints.
+
+    :param value: The interval value to validate.
+    :raises ValueError: ValueError: if value is int < 1,
+        or timedelta with fractional seconds, or < 1 second.
+    """
+    if value is None:
+        return
+    if isinstance(value, int):
+        if value < 1:
+            raise ValueError(
+                f"Interval must be at least 1 second, got {value} seconds",
+            )
+    else:
+        # For timedelta, check that it's at least 1 second
+        # and has no fractional seconds
+        total_seconds = value.total_seconds()
+        if total_seconds != int(total_seconds):
+            raise ValueError(
+                f"Fractional intervals are not supported, got {total_seconds} seconds",
+            )
+        if total_seconds < 1:
+            raise ValueError(
+                f"Interval must be at least 1 second, got {total_seconds} seconds",
+            )

tests/scheduler/test_interval_validation.py

@@ -1,92 +1,40 @@
-from datetime import datetime, timedelta
+from datetime import timedelta
 from typing import Union
 
 import pytest
 
-from taskiq.scheduler.scheduled_task import ScheduledTask
+from taskiq.scheduler.scheduled_task.validators import validate_interval_value
 
 
-def test_valid_interval_tasks() -> None:
-    """Test that valid interval tasks are created successfully."""
-    task1 = ScheduledTask(
-        task_name="test_task",
-        labels={},
-        args=[],
-        kwargs={},
-        interval=30,
-    )
-    assert task1.interval == 30
-
-    task2 = ScheduledTask(
-        task_name="test_task",
-        labels={},
-        args=[],
-        kwargs={},
-        interval=timedelta(minutes=5),
-    )
-    assert task2.interval == timedelta(minutes=5)
-
-    task3 = ScheduledTask(
-        task_name="test_task",
-        labels={},
-        args=[],
-        kwargs={},
-        interval=1,
-    )
-    assert task3.interval == 1
+@pytest.mark.parametrize(
+    "value",
+    [
+        None,
+        1,
+        5,
+        3600,
+        timedelta(seconds=1),
+        timedelta(seconds=5),
+        timedelta(hours=1, minutes=30),
+        timedelta(seconds=1, microseconds=0),
+    ],
+)
+def test_validate_interval_value_success(value: Union[int, timedelta, None]) -> None:
+    validate_interval_value(value)
 
 
 @pytest.mark.parametrize(
-    "interval",
+    "value",
     [
         0,
-        -5,
+        -1,
         timedelta(seconds=0),
-        timedelta(seconds=0.5),
-        timedelta(microseconds=500000),
-        timedelta(seconds=-1),
+        timedelta(milliseconds=999),
+        timedelta(seconds=1.5),
+        timedelta(seconds=0.999),
+        timedelta(seconds=1, microseconds=1),
     ],
 )
-def test_invalid_interval_tasks(interval: Union[int, timedelta]) -> None:
-    """Test that invalid interval tasks raise ValueError."""
+def test_validate_interval_value_fail(value: Union[int, timedelta, None]) -> None:
     with pytest.raises(ValueError):
-        ScheduledTask(
-            task_name="test_task",
-            labels={},
-            args=[],
-            kwargs={},
-            interval=interval,
-        )
-
-
-def test_interval_validation_with_other_schedule_types() -> None:
-    """Test that interval validation works with other schedule types."""
-    task1 = ScheduledTask(
-        task_name="test_task",
-        labels={},
-        args=[],
-        kwargs={},
-        cron="* * * * *",
-        interval=30,
-    )
-    assert task1.interval == 30
-
-    task2 = ScheduledTask(
-        task_name="test_task",
-        labels={},
-        args=[],
-        kwargs={},
-        time=datetime.now(),
-        interval=timedelta(minutes=5),
-    )
-    assert task2.interval == timedelta(minutes=5)
-
-    with pytest.raises(ValueError, match="Interval must be at least 1 second"):
-        ScheduledTask(
-            task_name="test_task",
-            labels={},
-            args=[],
-            kwargs={},
-            cron="* * * * *",
-            interval=0,
-        )
+        validate_interval_value(value)

tests/scheduler/test_scheduled_task.py

@@ -12,3 +12,15 @@ def test_scheduled_task_parameters() -> None:
             kwargs={},
             schedule_id="b",
         )
+
+
+def test_scheduled_task_interval() -> None:
+    with pytest.raises(ValueError):
+        ScheduledTask(
+            task_name="a",
+            labels={},
+            args=[],
+            kwargs={},
+            schedule_id="b",
+            interval=-1,
+        )