Skip to content

Commit

Permalink
Merge pull request #1169 from lsst-sqre/tickets/DM-47789
Browse files Browse the repository at this point in the history
DM-47789: Use Safir timedelta validation types
  • Loading branch information
rra authored Nov 26, 2024
2 parents fe13c91 + 9632de5 commit 860e7bf
Show file tree
Hide file tree
Showing 4 changed files with 8 additions and 96 deletions.
1 change: 1 addition & 0 deletions docs/documenteer.toml
Original file line number Diff line number Diff line change
Expand Up @@ -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"],
Expand Down
15 changes: 7 additions & 8 deletions src/gafaelfawr/models/kubernetes.py
Original file line number Diff line number Diff line change
Expand Up @@ -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

Expand All @@ -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__ = [
Expand Down Expand Up @@ -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")
)
Expand Down
68 changes: 0 additions & 68 deletions src/gafaelfawr/util.py
Original file line number Diff line number Diff line change
Expand Up @@ -6,28 +6,17 @@
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<weeks>\d+?)\s*(weeks|week|w))?\s*"
r"((?P<days>\d+?)\s*(days|day|d))?\s*"
r"((?P<hours>\d+?)\s*(hours|hour|hr|h))?\s*"
r"((?P<minutes>\d+?)\s*(minutes|minute|mins|min|m))?\s*"
r"((?P<seconds>\d+?)\s*(seconds|second|secs|sec|s))?$"
)
"""Regular expression pattern for a time duration."""

__all__ = [
"add_padding",
"base64_to_number",
"group_name_for_github_team",
"is_bot_user",
"normalize_ip_address",
"normalize_scopes",
"normalize_timedelta",
"number_to_base64",
"random_128_bits",
]
Expand Down Expand Up @@ -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.
Expand All @@ -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("=")
20 changes: 0 additions & 20 deletions tests/util_test.py
Original file line number Diff line number Diff line change
Expand Up @@ -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,
)

Expand Down Expand Up @@ -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"

0 comments on commit 860e7bf

Please sign in to comment.