Pydantic-friendly Timedelta wrapper

cadence_delta.py

# cadence_delta_test.py
from datetime import timedelta, datetime
from pydantic import BaseModel
from pydantic_core import core_schema
from pydantic import GetCoreSchemaHandler

import re

# Duration parsing
_TIME_UNITS = {
    "s": "seconds", "sec": "seconds", "secs": "seconds", "second": "seconds", "seconds": "seconds",
    "m": "minutes", "min": "minutes", "mins": "minutes", "minute": "minutes", "minutes": "minutes",
    "h": "hours", "hr": "hours", "hrs": "hours", "hour": "hours", "hours": "hours",
    "d": "days", "day": "days", "days": "days"
}

_DURATION_RE = re.compile(r"(?P<value>\d+(?:\.\d+)?)\s*(?P<unit>[a-zA-Z]+)", re.IGNORECASE)

def parse_duration_string(s: str) -> timedelta:
    s = s.strip()
    matches = _DURATION_RE.findall(s)
    if not matches:
        raise ValueError(f"Invalid duration string: '{s}'")

    kwargs = {}
    for value_str, unit_str in matches:
        unit_key = unit_str.lower()
        unit = _TIME_UNITS.get(unit_key)
        if not unit:
            raise ValueError(f"Unsupported time unit: '{unit_str}'")
        value = float(value_str)
        kwargs[unit] = kwargs.get(unit, 0) + value

    return timedelta(**kwargs)

def format_timedelta(delta: timedelta) -> str:
    total_seconds = int(delta.total_seconds())
    days, remainder = divmod(total_seconds, 86400)
    hours, remainder = divmod(remainder, 3600)
    minutes, seconds = divmod(remainder, 60)
    parts = []
    if days:
        parts.append(f"{days}d")
    if hours:
        parts.append(f"{hours}h")
    if minutes:
        parts.append(f"{minutes}m")
    if seconds or not parts:
        parts.append(f"{seconds}s")
    return " ".join(parts)

# Custom type
class CadenceDelta:
    def __init__(self, delta: timedelta):
        self.delta = delta

    def __str__(self):
        return format_timedelta(self.delta)

    def __repr__(self):
        return f"CadenceDelta({str(self)})"

    def __eq__(self, other):
        return isinstance(other, CadenceDelta) and self.delta == other.delta

    def as_timedelta(self) -> timedelta:
        return self.delta

    def is_expired(self, since: datetime, now: datetime | None = None) -> bool:
        now = now or datetime.now()
        return now - since >= self.delta

    @classmethod
    def __get_pydantic_core_schema__(cls, _source_type, handler: GetCoreSchemaHandler) -> core_schema.CoreSchema:
        return core_schema.json_or_python_schema(
            python_schema=core_schema.plain_validator_function(cls._validate),
            json_schema=core_schema.plain_validator_function(cls._validate),
            serialization=core_schema.plain_serializer_function_ser_schema(
                serializer=lambda instance, _: str(instance)
            )
        )

    @classmethod
    def _validate(cls, value):
        if isinstance(value, cls):
            return value
        if isinstance(value, timedelta):
            return cls(value)
        if isinstance(value, str):
            return cls(parse_duration_string(value))
        raise TypeError(f"Cannot convert {type(value)} to CadenceDelta")

# Pydantic model
class Config(BaseModel):
    cadence: CadenceDelta

# Test
if __name__ == "__main__":
    cfg = Config.model_validate({"cadence": "10 minutes"})
    print("cadence:", cfg.cadence)  # Expected: 10m
    print("timedelta:", cfg.cadence.as_timedelta())  # Expected: 0:10:00
    print("dump:", cfg.model_dump())  # {'cadence': '10m'}
    print("json:", cfg.model_dump_json())  # {"cadence": "10m"}