diff --git a/docs/documenteer.toml b/docs/documenteer.toml index c941af84..e38a35eb 100644 --- a/docs/documenteer.toml +++ b/docs/documenteer.toml @@ -48,6 +48,7 @@ nitpick_ignore = [ ["py:obj", "fastapi.routing.APIRoute"], ["py:class", "kubernetes_asyncio.client.api_client.ApiClient"], ["py:class", "pydantic.functional_serializers.PlainSerializer"], + ["py:class", "pydantic.functional_validators.AfterValidator"], ["py:class", "pydantic.functional_validators.BeforeValidator"], ["py:class", "pydantic.main.BaseModel"], ["py:class", "pydantic.networks.UrlConstraints"], diff --git a/src/gafaelfawr/models/kubernetes.py b/src/gafaelfawr/models/kubernetes.py index 78d746f9..037b78f8 100644 --- a/src/gafaelfawr/models/kubernetes.py +++ b/src/gafaelfawr/models/kubernetes.py @@ -4,7 +4,7 @@ from abc import ABCMeta, abstractmethod from dataclasses import dataclass, field -from datetime import datetime, timedelta +from datetime import datetime from enum import Enum from typing import Literal, Self, override @@ -26,9 +26,12 @@ ) from pydantic.alias_generators import to_camel from safir.datetime import current_datetime -from safir.pydantic import to_camel_case, validate_exactly_one_of +from safir.pydantic import ( + SecondsTimedelta, + to_camel_case, + validate_exactly_one_of, +) -from ..util import normalize_timedelta from .auth import AuthType, Satisfy __all__ = [ @@ -156,16 +159,12 @@ class GafaelfawrIngressDelegate(BaseModel): internal: GafaelfawrIngressDelegateInternal | None = None """Configuration for a delegated internal token.""" - minimum_lifetime: timedelta | None = None + minimum_lifetime: SecondsTimedelta | None = None """The minimum lifetime of the delegated token.""" use_authorization: bool = False """Whether to put the delegated token in the ``Authorization`` header.""" - _normalize_minimum_lifetime = field_validator( - "minimum_lifetime", mode="before" - )(normalize_timedelta) - _validate_type = model_validator(mode="after")( validate_exactly_one_of("notebook", "internal") ) diff --git a/src/gafaelfawr/util.py b/src/gafaelfawr/util.py index 9f78b23f..90306be0 100644 --- a/src/gafaelfawr/util.py +++ b/src/gafaelfawr/util.py @@ -6,20 +6,10 @@ import hashlib import os import re -from datetime import timedelta from ipaddress import IPv4Address, IPv6Address from .constants import BOT_USERNAME_REGEX -_TIMEDELTA_PATTERN = re.compile( - r"((?P\d+?)\s*(weeks|week|w))?\s*" - r"((?P\d+?)\s*(days|day|d))?\s*" - r"((?P\d+?)\s*(hours|hour|hr|h))?\s*" - r"((?P\d+?)\s*(minutes|minute|mins|min|m))?\s*" - r"((?P\d+?)\s*(seconds|second|secs|sec|s))?$" -) -"""Regular expression pattern for a time duration.""" - __all__ = [ "add_padding", "base64_to_number", @@ -27,7 +17,6 @@ "is_bot_user", "normalize_ip_address", "normalize_scopes", - "normalize_timedelta", "number_to_base64", "random_128_bits", ] @@ -168,32 +157,6 @@ def normalize_scopes(v: str | list[str] | None) -> list[str] | None: return v -def normalize_timedelta(v: int | timedelta | None) -> timedelta | None: - """Pydantic validator for timedelta fields. - - The only reason to use this validator over Pydantic's built-in behavior is - to ensure that ISO time durations are rejected and only an integer number - of seconds is supported. - - Parameters - ---------- - v - The field representing a duration, in seconds. - - Returns - ------- - datetime.timedelta or None - The corresponding `datetime.timedelta` or `None` if the input was - `None`. - """ - if v is None or isinstance(v, timedelta): - return v - elif isinstance(v, int): - return timedelta(seconds=v) - else: - raise ValueError("invalid timedelta (should be in seconds)") - - def number_to_base64(data: int) -> bytes: """Convert an integer to base64-encoded bytes in big endian order. @@ -218,37 +181,6 @@ def number_to_base64(data: int) -> bytes: return base64.urlsafe_b64encode(data_as_bytes).rstrip(b"=") -def parse_timedelta(text: str) -> timedelta: - """Parse a string into a `datetime.timedelta`. - - This function can be used as a before-mode validator for Pydantic, - replacing Pydantic's default ISO 8601 duration support. Expects a string - consisting of one or more sequences of numbers and duration abbreviations, - separated by optional whitespace. The supported abbreviations are: - - - Week: ``weeks``, ``week``, ``w`` - - Day: ``days``, ``day``, ``d`` - - Hour: ``hours``, ``hour``, ``hr``, ``h`` - - Minute: ``minutes``, ``minute``, ``mins``, ``min``, ``m`` - - Second: ``seconds``, ``second``, ``secs``, ``sec``, ``s`` - - Parameters - ---------- - text - Input string. - - Returns - ------- - timedelta - Converted `datetime.timedelta`. - """ - m = _TIMEDELTA_PATTERN.match(text.strip()) - if m is None: - raise ValueError(f"Could not parse {text!r} as a time duration") - td_args = {k: int(v) for k, v in m.groupdict().items() if v is not None} - return timedelta(**td_args) - - def random_128_bits() -> str: """Generate random 128 bits encoded in base64 without padding.""" return base64.urlsafe_b64encode(os.urandom(16)).decode().rstrip("=") diff --git a/tests/util_test.py b/tests/util_test.py index 9f5dd768..d6c007fa 100644 --- a/tests/util_test.py +++ b/tests/util_test.py @@ -2,17 +2,11 @@ from __future__ import annotations -from datetime import timedelta - -import pytest -from pydantic import BaseModel, field_validator - from gafaelfawr.keypair import RSAKeyPair from gafaelfawr.util import ( add_padding, base64_to_number, is_bot_user, - normalize_timedelta, number_to_base64, ) @@ -52,20 +46,6 @@ def test_is_bot_user() -> None: assert not is_bot_user("bot-in!valid") -def test_normalize_timedelta() -> None: - class TestModel(BaseModel): - delta: timedelta | None - - _val = field_validator("delta", mode="before")(normalize_timedelta) - - assert TestModel(delta=None).delta is None - model = TestModel(delta=10) # type: ignore[arg-type] - assert model.delta == timedelta(seconds=10) - - with pytest.raises(ValueError, match="invalid timedelta"): - TestModel(delta="not an int") # type: ignore[arg-type] - - def test_number_to_base64() -> None: assert number_to_base64(0) == b"AA" assert number_to_base64(65537) == b"AQAB"