Convert datetime and timedelta to Safir types

Convert the Gafaelfawr `Timestamp` data type to be based on the
Safir `UtcDatetime` instead of using the `normalize_datetime`
function, which will hopefully be deprecated in the future.
Convert some other `datetime` model fields to `UtcDatetime`.

Convert one remaining `timedelta` query parameter to use
`SecondsTimedelta` so that Pydantic will do the validation.

docs/documenteer.toml

@@ -30,9 +30,10 @@ nitpick_ignore = [
     # Ignore missing cross-references for modules that don't provide
     # intersphinx. The documentation itself should use double-quotes instead
     # of single-quotes to not generate a reference, but automatic references
-    # are generated from the type signatures and can't be avoided. These are
-    # intentionally listed specifically because I've caught documentation bugs
-    # by having Sphinx complain about a new symbol.
+    # are generated from the type signatures and can't be avoided.
+    #
+    # These are listed specifically rather than with regexesbecause I've
+    # caught documentation bugs by having Sphinx complain about a new symbol.
     ["py:class", "dataclasses_avroschema.pydantic.main.AvroBaseModel"],
     ["py:class", "dataclasses_avroschema.main.AvroModel"],
     ["py:class", "google.cloud.firestore_v1.async_client.AsyncClient"],
@@ -42,10 +43,6 @@ nitpick_ignore = [
     ["py:class", "fastapi.params.Depends"],
     ["py:class", "fastapi.routing.APIRoute"],
     ["py:class", "httpx.AsyncClient"],
-    ["py:exc", "fastapi.HTTPException"],
-    ["py:exc", "fastapi.exceptions.RequestValidationError"],
-    ["py:exc", "httpx.HTTPError"],
-    ["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"],
@@ -64,7 +61,11 @@ nitpick_ignore = [
     ["py:class", "starlette.responses.Response"],
     ["py:class", "starlette.routing.Route"],
     ["py:class", "starlette.routing.BaseRoute"],
+    ["py:exc", "fastapi.HTTPException"],
+    ["py:exc", "fastapi.exceptions.RequestValidationError"],
+    ["py:exc", "httpx.HTTPError"],
     ["py:exc", "starlette.exceptions.HTTPException"],
+    ["py:obj", "fastapi.routing.APIRoute"],
     # Broken links created by autodoc_pydantic.
     ["py:class", "lambda"],
     ["py:class", "safir.pydantic._validators.normalize_datetime"],

src/gafaelfawr/handlers/api.py

@@ -6,7 +6,6 @@
 models.
 """
 
-from datetime import datetime
 from typing import Annotated, Any
 from urllib.parse import quote
 
@@ -20,6 +19,7 @@
     status,
 )
 from safir.models import ErrorLocation, ErrorModel
+from safir.pydantic import UtcDatetime
 from safir.slack.webhook import SlackRouteErrorHandler
 
 from ..constants import ACTOR_REGEX, CURSOR_REGEX, USERNAME_REGEX
@@ -180,15 +180,15 @@ async def get_admin_token_change_history(
         ),
     ] = None,
     since: Annotated[
-        datetime | None,
+        UtcDatetime | None,
         Query(
             title="Not before",
             description="Only show entries at or after this time",
             examples=["2021-03-05T14:59:52Z"],
         ),
     ] = None,
     until: Annotated[
-        datetime | None,
+        UtcDatetime | None,
         Query(
             title="Not after",
             description="Only show entries before or at this time",
@@ -430,15 +430,15 @@ async def get_user_token_change_history(
         ),
     ] = None,
     since: Annotated[
-        datetime | None,
+        UtcDatetime | None,
         Query(
             title="Not before",
             description="Only show entries at or after this time",
             examples=["2021-03-05T14:59:52Z"],
         ),
     ] = None,
     until: Annotated[
-        datetime | None,
+        UtcDatetime | None,
         Query(
             title="Not after",
             description="Only show entries before or at this time",

src/gafaelfawr/handlers/ingress.py

@@ -16,6 +16,7 @@
 from fastapi import APIRouter, Depends, Header, HTTPException, Query, Response
 from safir.datetime import current_datetime
 from safir.models import ErrorModel
+from safir.pydantic import SecondsTimedelta
 from safir.slack.webhook import SlackRouteErrorHandler
 
 from ..auth import (
@@ -139,15 +140,15 @@ def auth_config(
         ),
     ] = None,
     minimum_lifetime: Annotated[
-        int | None,
+        SecondsTimedelta | None,
         Query(
             title="Required minimum lifetime",
             description=(
                 "Force reauthentication if the delegated token (internal or"
                 " notebook) would have a shorter lifetime, in seconds, than"
                 " this parameter."
             ),
-            ge=MINIMUM_LIFETIME.total_seconds(),
+            ge=MINIMUM_LIFETIME,
             examples=[86400],
         ),
     ] = None,
@@ -266,16 +267,13 @@ def auth_config(
         delegate_scopes = {s.strip() for s in delegate_scope.split(",")}
     else:
         delegate_scopes = set()
-    lifetime = None
-    if minimum_lifetime:
-        lifetime = timedelta(seconds=minimum_lifetime)
-    elif not minimum_lifetime and (notebook or delegate_to):
-        lifetime = MINIMUM_LIFETIME
+    if not minimum_lifetime and (notebook or delegate_to):
+        minimum_lifetime = MINIMUM_LIFETIME
     return AuthConfig(
         auth_type=auth_type,
         delegate_scopes=delegate_scopes,
         delegate_to=delegate_to,
-        minimum_lifetime=lifetime,
+        minimum_lifetime=minimum_lifetime,
         only_services=set(only_service) if only_service else None,
         notebook=notebook,
         satisfy=satisfy,

src/gafaelfawr/models/kubernetes.py

@@ -4,7 +4,6 @@
 
 from abc import ABCMeta, abstractmethod
 from dataclasses import dataclass, field
-from datetime import datetime
 from enum import Enum
 from typing import Literal, Self, override
 
@@ -28,6 +27,7 @@
 from safir.datetime import current_datetime
 from safir.pydantic import (
     SecondsTimedelta,
+    UtcDatetime,
     to_camel_case,
     validate_exactly_one_of,
 )
@@ -614,7 +614,7 @@ class KubernetesResourceStatus:
     """Represents the processing status of a Kubernetes resource.
 
     This is returned as the result of the Kopf_ operator handlers for changes
-    to a Kubernetes resource.  Kopf will then put this information into the
+    to a Kubernetes resource. Kopf will then put this information into the
     ``status`` field of the GafaelfawrServiceToken object.
     """
 
@@ -627,7 +627,7 @@ class KubernetesResourceStatus:
     reason: StatusReason
     """Reason for the status update."""
 
-    timestamp: datetime = field(default_factory=current_datetime)
+    timestamp: UtcDatetime = field(default_factory=current_datetime)
     """Time of the status event."""
 
     @classmethod
@@ -637,14 +637,14 @@ def failure(cls, resource: KubernetesResource, message: str) -> Self:
         Parameters
         ----------
         service_token
-            The object being processed.
+            Object being processed.
         message
-            The error message for the failure.
+            Error message for the failure.
 
         Returns
         -------
         KubernetesResourceStatus
-            The corresponding status object.
+            Corresponding status object.
         """
         return cls(
             message=message,

src/gafaelfawr/models/token.py

@@ -2,11 +2,11 @@
 
 from __future__ import annotations
 
-from datetime import datetime
 from typing import Any, Self
 
 from pydantic import BaseModel, Field, ValidationInfo, field_validator
 from safir.datetime import current_datetime
+from safir.pydantic import UtcDatetime
 
 from ..constants import USERNAME_REGEX
 from ..exceptions import InvalidTokenError
@@ -318,7 +318,6 @@ def bootstrap_token(cls) -> Self:
             username="<bootstrap>",
             token_type=TokenType.service,
             scopes=["admin:token"],
-            created=current_datetime(),
         )
 
     @classmethod
@@ -338,7 +337,6 @@ def internal_token(cls) -> Self:
             username="<internal>",
             token_type=TokenType.service,
             scopes=["admin:token"],
-            created=current_datetime(),
         )
 
 
@@ -391,7 +389,7 @@ class AdminTokenRequest(BaseModel):
         examples=[["read:all"]],
     )
 
-    expires: datetime | None = Field(
+    expires: UtcDatetime | None = Field(
         None,
         title="Token expiration",
         description=(
@@ -497,7 +495,7 @@ class UserTokenRequest(BaseModel):
         examples=[["read:all"]],
     )
 
-    expires: datetime | None = Field(
+    expires: UtcDatetime | None = Field(
         None,
         title="Expiration time",
         description="Expiration timestamp of the token in seconds since epoch",
@@ -524,7 +522,7 @@ class UserTokenModifyRequest(BaseModel):
         None, title="Token scopes", examples=[["read:all"]]
     )
 
-    expires: datetime | None = Field(
+    expires: UtcDatetime | None = Field(
         None,
         title="Expiration time",
         description=(

src/gafaelfawr/models/userinfo.py

@@ -2,11 +2,10 @@
 
 from __future__ import annotations
 
-from datetime import datetime
-
-from pydantic import BaseModel, Field, field_serializer
+from pydantic import BaseModel, Field
 
 from ..constants import GROUPNAME_REGEX
+from ..pydantic import Timestamp
 
 __all__ = [
     "CADCUserInfo",
@@ -24,7 +23,7 @@ class CADCUserInfo(BaseModel):
     support with the OpenID Connect support.
     """
 
-    exp: datetime | None = Field(
+    exp: Timestamp | None = Field(
         None,
         title="Expiration time",
         description=(
@@ -52,10 +51,6 @@ class CADCUserInfo(BaseModel):
         examples=["someuser"],
     )
 
-    @field_serializer("exp")
-    def _serialize_datetime(self, time: datetime | None) -> int | None:
-        return int(time.timestamp()) if time is not None else None
-
 
 class Group(BaseModel):
     """Information about a single group."""

src/gafaelfawr/pydantic.py

@@ -2,18 +2,16 @@
 
 from __future__ import annotations
 
-from datetime import datetime
 from typing import Annotated, TypeAlias
 
-from pydantic import BeforeValidator, PlainSerializer
-from safir.pydantic import normalize_datetime
+from pydantic import PlainSerializer
+from safir.pydantic import UtcDatetime
 
 __all__ = ["Timestamp"]
 
 
 Timestamp: TypeAlias = Annotated[
-    datetime,
-    BeforeValidator(normalize_datetime),
+    UtcDatetime,
     PlainSerializer(lambda t: int(t.timestamp()), return_type=int),
 ]
-"""Type for a `datetime` field that only accepts seconds since epoch."""
+"""Type for a `datetime` field that serializes to seconds since epoch."""