diff --git a/pydantic_extra_types/cron.py b/pydantic_extra_types/cron.py new file mode 100644 index 0000000..e6390d5 --- /dev/null +++ b/pydantic_extra_types/cron.py @@ -0,0 +1,134 @@ +"""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/). + ## Examples + ```python + from pydantic import BaseModel + from pydantic_extra_types.cron import CronStr + + class Schedule(BaseModel): + cron: CronStr + + schedule = Schedule(cron="*/5 * * * *") + print(schedule.cron) + >> */5 * * * * + print(schedule.cron.minute) + >> */5 + print(schedule.cron.next_run) + >> 2025-10-07T22:40:00+00:00 + ``` + """ + + 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() diff --git a/pyproject.toml b/pyproject.toml index 023c7a2..276123f 100644 --- a/pyproject.toml +++ b/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 = [ diff --git a/tests/test_cron.py b/tests/test_cron.py new file mode 100644 index 0000000..e97e7aa --- /dev/null +++ b/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 * * *' diff --git a/tests/test_json_schema.py b/tests/test_json_schema.py index 815ac53..27e72a2 100644 --- a/tests/test_json_schema.py +++ b/tests/test_json_schema.py @@ -10,6 +10,7 @@ from pydantic_extra_types.color import Color from pydantic_extra_types.coordinate import Coordinate, Latitude, Longitude from pydantic_extra_types.country import CountryAlpha2, CountryAlpha3, CountryNumericCode, CountryShortName +from pydantic_extra_types.cron import CronStr from pydantic_extra_types.currency_code import ISO4217, Currency from pydantic_extra_types.domain import DomainStr from pydantic_extra_types.isbn import ISBN @@ -93,6 +94,20 @@ 'type': 'object', }, ), + ( + CronStr, + { + 'title': 'Model', + 'type': 'object', + 'properties': { + 'x': { + 'title': 'X', + 'type': 'string', + } + }, + 'required': ['x'], + }, + ), ( CountryAlpha3, { diff --git a/uv.lock b/uv.lock index 3b9c24b..0aff7f6 100644 --- a/uv.lock +++ b/uv.lock @@ -205,6 +205,18 @@ toml = [ { name = "tomli", marker = "python_full_version >= '3.9' and python_full_version <= '3.11'" }, ] +[[package]] +name = "cron-converter" +version = "1.2.2" +source = { registry = "https://pypi.org/simple" } +dependencies = [ + { name = "python-dateutil" }, +] +sdist = { url = "https://files.pythonhosted.org/packages/c7/b9/744734ae853c43f8b71510402e0ab0106ba96a741113f9e26e89fde40736/cron_converter-1.2.2.tar.gz", hash = "sha256:b987525ddf7d5ad28286620622f00dde61c73833d1f05c332a26c389a9c512c3", size = 14509, upload-time = "2025-07-21T09:25:00.01Z" } +wheels = [ + { url = "https://files.pythonhosted.org/packages/62/f8/1698a37dd13fc120a96a062c10dba11cade6ebb71b68a14ee177032b4b61/cron_converter-1.2.2-py3-none-any.whl", hash = "sha256:a31c71223cc71f07f9af2533af50c4cf1b910a85a730dd06a00aed053ea250fe", size = 13434, upload-time = "2025-07-21T09:24:58.638Z" }, +] + [[package]] name = "dirty-equals" version = "0.9.0" @@ -878,6 +890,7 @@ dependencies = [ [package.optional-dependencies] all = [ + { name = "cron-converter" }, { name = "pendulum", version = "3.0.0", source = { registry = "https://pypi.org/simple" }, marker = "python_full_version < '3.9'" }, { name = "pendulum", version = "3.1.0", source = { registry = "https://pypi.org/simple" }, marker = "python_full_version >= '3.9'" }, { name = "phonenumbers" }, @@ -890,6 +903,9 @@ all = [ { name = "semver" }, { name = "tzdata" }, ] +cron = [ + { name = "cron-converter" }, +] pendulum = [ { name = "pendulum", version = "3.0.0", source = { registry = "https://pypi.org/simple" }, marker = "python_full_version < '3.9'" }, { name = "pendulum", version = "3.1.0", source = { registry = "https://pypi.org/simple" }, marker = "python_full_version >= '3.9'" }, @@ -940,6 +956,8 @@ lint = [ [package.metadata] requires-dist = [ + { name = "cron-converter", marker = "extra == 'all'", specifier = ">=1.2.2" }, + { name = "cron-converter", marker = "extra == 'cron'", specifier = ">=1.2.2" }, { name = "pendulum", marker = "extra == 'all'", specifier = ">=3.0.0,<4.0.0" }, { name = "pendulum", marker = "extra == 'pendulum'", specifier = ">=3.0.0,<4.0.0" }, { name = "phonenumbers", marker = "extra == 'all'", specifier = ">=8,<10" }, @@ -959,7 +977,7 @@ requires-dist = [ { name = "typing-extensions" }, { name = "tzdata", marker = "extra == 'all'", specifier = ">=2024.1" }, ] -provides-extras = ["all", "pendulum", "phonenumbers", "pycountry", "python-ulid", "semver"] +provides-extras = ["all", "cron", "pendulum", "phonenumbers", "pycountry", "python-ulid", "semver"] [package.metadata.requires-dev] dev = [