Added cron type

pydantic_extra_types/cron.py

@@ -0,0 +1,117 @@
+"""The `pydantic_extra_types.cron` module provides the [`CronStr`][pydantic_extra_types.cron.CronStr] data type."""
+
+from __future__ import annotations
+
+from datetime import datetime
+from typing import TYPE_CHECKING, Any, ClassVar, Protocol, cast
+
+try:
+    from cron_converter import Cron  # type: ignore[import-untyped]
+except ModuleNotFoundError as e:  # pragma: no cover
+    raise RuntimeError(
+        'The `cron` module requires "cron-converter" to be installed. You can install it with "pip install cron-converter".'
+    ) from e
+from pydantic import GetCoreSchemaHandler
+from pydantic_core import PydanticCustomError, core_schema
+
+if TYPE_CHECKING:
+    from cron_converter.sub_modules.seeker import Seeker as CronSeeker  # type: ignore[import-untyped]
+else:
+
+    class CronSeeker(Protocol):
+        def next(self) -> datetime: ...
+
+
+class CronStr(str):
+    """A cron expression validated via [`cron-converter`](https://pypi.org/project/cron-converter/)."""
+
+    strip_whitespace: ClassVar[bool] = True
+    """Whether to strip surrounding whitespace from the input value."""
+    _component_names: ClassVar[tuple[str, ...]] = (
+        'minute',
+        'hour',
+        'day_of_the_month',
+        'month',
+        'day_of_the_week',
+    )
+    """Expected cron expression components in the order enforced by `cron-converter`."""
+
+    minute: str
+    hour: str
+    day_of_the_month: str
+    month: str
+    day_of_the_week: str
+    cron_obj: Cron
+
+    def __new__(cls, cron_expression: str, *, _cron: Cron | None = None) -> CronStr:
+        if _cron is None:
+            cron_expression, cron_obj = cls._validate(cron_expression)
+        else:
+            cron_obj = _cron
+            cron_expression = cron_obj.to_string()
+
+        obj = super().__new__(cls, cron_expression)
+        obj._apply_cron(cron_obj)
+        return obj
+
+    def _apply_cron(self, cron_obj: Cron) -> None:
+        self.cron_obj = cron_obj
+        self.minute, self.hour, self.day_of_the_month, self.month, self.day_of_the_week = str(self).split()
+
+    @classmethod
+    def _validate(cls, value: Any) -> tuple[str, Cron]:
+        if not isinstance(value, str):
+            raise PydanticCustomError('cron_str_type', 'Cron expression must be a string')
+
+        cron_expression = value.strip()
+        if not cron_expression:
+            raise PydanticCustomError('cron_str_empty', 'Cron expression must not be empty')
+
+        parts = cron_expression.split()
+        if len(parts) != len(cls._component_names):
+            parts_list = ', '.join(cls._component_names)
+            raise PydanticCustomError(
+                'cron_str_components',
+                f'Cron expression must contain {len(cls._component_names)} space separated components: {parts_list}',
+            )
+
+        try:
+            cron_obj = Cron(cron_expression)
+        except (TypeError, ValueError) as exc:
+            raise PydanticCustomError('cron_str_invalid', str(exc)) from exc
+
+        # `cron-converter` may normalise components (e.g. remove duplicate spaces),
+        # so we reuse its canonical representation.
+        return cron_obj.to_string(), cron_obj
+
+    @classmethod
+    def validate(cls, __input_value: Any, _: core_schema.ValidationInfo) -> CronStr:
+        cron_expression, cron_obj = cls._validate(__input_value)
+        return cls(cron_expression, _cron=cron_obj)
+
+    @classmethod
+    def __get_pydantic_core_schema__(cls, source_type: Any, handler: GetCoreSchemaHandler) -> core_schema.CoreSchema:
+        return core_schema.with_info_after_validator_function(
+            cls.validate,
+            core_schema.str_schema(strip_whitespace=cls.strip_whitespace),
+        )
+
+    @classmethod
+    def __get_pydantic_json_schema__(
+        cls, schema: core_schema.CoreSchema, handler: GetCoreSchemaHandler
+    ) -> dict[str, Any]:
+        return dict(handler(schema))
+
+    def schedule(self, start_date: datetime | None = None, timezone_str: str | None = None) -> CronSeeker:
+        """Return the iterator produced by `cron-converter` for this expression."""
+        return cast(CronSeeker, self.cron_obj.schedule(start_date=start_date, timezone_str=timezone_str))
+
+    def next_after(self, start_date: datetime | None = None, timezone_str: str | None = None) -> datetime:
+        """Return the first run datetime after `start_date` (or now if omitted)."""
+        seeker = self.schedule(start_date=start_date, timezone_str=timezone_str)
+        return cast(datetime, seeker.next())
+
+    @property
+    def next_run(self) -> str:
+        """Return the next run as an ISO formatted string (shortcut for backwards compatibility)."""
+        return self.next_after().isoformat()

pyproject.toml

@@ -54,6 +54,7 @@ all = [
     'pytz>=2024.1',
     'semver~=3.0.2',
     'tzdata>=2024.1',
+    "cron-converter>=1.2.2",
 ]
 phonenumbers = ['phonenumbers>=8,<10']
 pycountry = ['pycountry>=23']
@@ -63,6 +64,7 @@ python_ulid = [
     'python-ulid>=1,<4; python_version>="3.9"',
 ]
 pendulum = ['pendulum>=3.0.0,<4.0.0']
+cron = ['cron-converter>=1.2.2']
 
 [dependency-groups]
 dev = [

tests/test_cron.py

@@ -0,0 +1,49 @@
+from datetime import datetime, timezone
+
+import pytest
+from cron_converter import Cron
+from pydantic import BaseModel, ValidationError
+
+from pydantic_extra_types.cron import CronStr
+
+
+class CronModel(BaseModel):
+    cron: CronStr
+
+
+def test_cron_str_is_validated_via_model() -> None:
+    model = CronModel(cron='*/5 0 * * 1-5')
+    cron_value = model.cron
+
+    assert isinstance(cron_value, CronStr)
+    assert cron_value.minute == '*/5'
+    assert cron_value.hour == '0'
+    assert cron_value.day_of_the_month == '*'
+    assert cron_value.month == '*'
+    assert cron_value.day_of_the_week == '1-5'
+    assert isinstance(cron_value.cron_obj, Cron)
+
+
+def test_cron_str_rejects_invalid_components() -> None:
+    with pytest.raises(ValidationError) as exc:
+        CronModel(cron='* * * *')
+    assert 'Cron expression must contain 5 space separated components' in str(exc.value)
+
+
+def test_cron_str_rejects_invalid_expression() -> None:
+    with pytest.raises(ValidationError) as exc:
+        CronModel(cron='60 0 * * *')
+    assert "Value 60 out of range for 'minute'" in str(exc.value)
+
+
+def test_cron_str_next_after() -> None:
+    cron_value = CronStr('15 8 * * 1-5')
+    next_run = cron_value.next_after(datetime(2024, 1, 1, 7, 0))
+    assert next_run == datetime(2024, 1, 1, 8, 15)
+    current_year = datetime.now(timezone.utc).year
+    assert cron_value.next_run.startswith(str(current_year))  # sanity check for property access
+
+
+def test_cron_str_strips_whitespace() -> None:
+    cron_value = CronStr(' 0 12 * * * ')
+    assert str(cron_value) == '0 12 * * *'