Coverage for .venv/lib/python3.13/site-packages/litellm/proxy/common_utils/timezone_utils.py: 63%
41 statements
« prev ^ index » next coverage.py v7.15.2, created at 2026-10-10 12:01 +0000
« prev ^ index » next coverage.py v7.15.2, created at 2026-10-10 12:01 +0000
1from datetime import datetime, time, timezone
3from pydantic import BaseModel, ConfigDict
5import litellm
6from litellm.litellm_core_utils.duration_parser import get_next_standardized_reset_time
9class BudgetResetSettings(BaseModel):
10 """Immutable, validated settings that govern when budgets reset.
12 Parsed once from `litellm_settings` and injected into consumers (the reset
13 job, management endpoints) so reset times never depend on reaching into
14 module-level globals at call time.
15 """
17 model_config = ConfigDict(frozen=True)
19 timezone: str = "UTC"
20 reset_time_of_day: time = time(0, 0)
23def parse_budget_reset_time(raw: object) -> time:
24 """Parse a `budget_reset_time` config value (e.g. "12:00") into a `time`.
26 Falls back to midnight when unset; raises a clear error on a malformed value
27 so a bad config fails loudly at startup instead of silently resetting at midnight.
28 """
29 if raw is None or raw == "": 29 ↛ 31line 29 didn't jump to line 31 because the condition on line 29 was always true
30 return time(0, 0)
31 if not isinstance(raw, str):
32 raise ValueError(f"Invalid budget_reset_time {raw!r}; must be a quoted 24-hour 'HH:MM' string, e.g. \"12:00\"")
33 for fmt in ("%H:%M", "%H:%M:%S"):
34 try:
35 parsed = datetime.strptime(raw, fmt)
36 return time(hour=parsed.hour, minute=parsed.minute, second=parsed.second)
37 except ValueError:
38 continue
39 raise ValueError(
40 f"Invalid budget_reset_time {raw!r}; expected a 24-hour 'HH:MM' or 'HH:MM:SS' string, e.g. \"12:00\""
41 )
44def get_budget_reset_timezone() -> str:
45 """
46 Get the budget reset timezone from litellm_settings.
47 Falls back to UTC if not specified.
49 litellm_settings values are set as attributes on the litellm module
50 by proxy_server.py at startup (via setattr(litellm, key, value)).
51 """
52 return getattr(litellm, "timezone", None) or "UTC"
55def get_budget_reset_settings() -> BudgetResetSettings:
56 """Build validated reset settings from litellm_settings. Raises on a malformed
57 `budget_reset_time`, which lets the proxy fail fast at startup."""
58 return BudgetResetSettings(
59 timezone=get_budget_reset_timezone(),
60 reset_time_of_day=parse_budget_reset_time(getattr(litellm, "budget_reset_time", None)),
61 )
64def compute_budget_reset_at(budget_duration: str, settings: BudgetResetSettings) -> datetime:
65 """Compute the next reset time for a budget duration using injected settings."""
66 return get_next_standardized_reset_time(
67 duration=budget_duration,
68 current_time=datetime.now(timezone.utc),
69 timezone_str=settings.timezone,
70 reset_time_of_day=settings.reset_time_of_day,
71 )
74def get_budget_reset_time(budget_duration: str) -> datetime:
75 """Get the budget reset time using the globally-configured timezone and reset time.
77 Thin wrapper over `compute_budget_reset_at` for callers that don't yet receive
78 `BudgetResetSettings` by injection (creation/update endpoints, startup backfill).
79 """
80 return compute_budget_reset_at(budget_duration, get_budget_reset_settings())
83def _is_persistable_budget_duration(budget_duration: str) -> bool:
84 from litellm.litellm_core_utils.duration_parser import duration_in_seconds
86 try:
87 if duration_in_seconds(budget_duration) <= 0: 87 ↛ anywhereline 87 didn't jump anywhere: it always raised an exception.
88 return False
89 get_budget_reset_time(budget_duration=budget_duration)
90 except (ValueError, OverflowError):
91 return False
92 return True
95def budget_duration_error(budget_duration: str | None) -> str | None:
96 """Why `budget_duration` cannot be persisted, or None when it is usable.
98 A non-positive duration resolves to a reset time of "now", which leaves the row
99 permanently due: the reset job re-reads it every tick and, once enough of them
100 exist, they fill each batch and starve every other tenant's reset.
101 """
102 if budget_duration is None or _is_persistable_budget_duration(budget_duration):
103 return None
104 return f"Invalid budget_duration '{budget_duration}'. Use a format like '1h', '24h', '7d', or '30d'."