zl-q
79 lines
2.2 KiB
Python
79 lines
2.2 KiB
Python
from __future__ import annotations
|
|
|
|
from datetime import datetime
|
|
import re
|
|
from typing import Any
|
|
from zoneinfo import ZoneInfo, ZoneInfoNotFoundError
|
|
|
|
from pydantic import (
|
|
BaseModel,
|
|
ConfigDict,
|
|
Field,
|
|
StrictInt,
|
|
ValidationError,
|
|
field_validator,
|
|
)
|
|
|
|
_RFC3339_WITH_TZ_PATTERN = re.compile(
|
|
r"^\d{4}-\d{2}-\d{2}T\d{2}:\d{2}:\d{2}(?:\.\d+)?(?:Z|[+-]\d{2}:\d{2})$"
|
|
)
|
|
|
|
|
|
class ClientTimeContext(BaseModel):
|
|
model_config = ConfigDict(extra="forbid")
|
|
|
|
device_timezone: str = Field(
|
|
...,
|
|
description="IANA timezone from client device, e.g. America/Los_Angeles.",
|
|
)
|
|
client_now_iso: str = Field(
|
|
...,
|
|
description="RFC3339 datetime with timezone offset from client device.",
|
|
)
|
|
client_epoch_ms: StrictInt = Field(
|
|
...,
|
|
ge=0,
|
|
description="Unix epoch milliseconds from client device.",
|
|
)
|
|
|
|
@field_validator("device_timezone")
|
|
@classmethod
|
|
def validate_device_timezone(cls, value: str) -> str:
|
|
try:
|
|
ZoneInfo(value)
|
|
except ZoneInfoNotFoundError as exc:
|
|
raise ValueError("invalid client_time.device_timezone") from exc
|
|
return value
|
|
|
|
@field_validator("client_now_iso")
|
|
@classmethod
|
|
def validate_client_now_iso(cls, value: str) -> str:
|
|
if not _RFC3339_WITH_TZ_PATTERN.fullmatch(value):
|
|
raise ValueError("invalid client_time.client_now_iso")
|
|
normalized = value.replace("Z", "+00:00")
|
|
try:
|
|
parsed = datetime.fromisoformat(normalized)
|
|
except ValueError as exc:
|
|
raise ValueError("invalid client_time.client_now_iso") from exc
|
|
if parsed.tzinfo is None:
|
|
raise ValueError("invalid client_time.client_now_iso")
|
|
return value
|
|
|
|
|
|
class ForwardedPropsPayload(BaseModel):
|
|
model_config = ConfigDict(extra="forbid")
|
|
|
|
client_time: ClientTimeContext | None = None
|
|
|
|
|
|
def parse_forwarded_props_client_time(
|
|
forwarded_props: Any,
|
|
) -> ClientTimeContext | None:
|
|
if not isinstance(forwarded_props, dict):
|
|
return None
|
|
try:
|
|
payload = ForwardedPropsPayload.model_validate(forwarded_props)
|
|
except ValidationError as exc:
|
|
raise ValueError("invalid RunAgentInput.forwardedProps") from exc
|
|
return payload.client_time
|