| """Health-related domain models.""" |
|
|
| from __future__ import annotations |
|
|
| from datetime import datetime, timezone |
| from typing import Dict, List, Optional |
|
|
| from pydantic import BaseModel, Field |
|
|
|
|
| class ProviderHealthSnapshot(BaseModel): |
| """Per-provider health snapshot used by the circuit breaker + API.""" |
| name: str |
| healthy: bool |
| consecutive_failures: int = 0 |
| last_success: Optional[datetime] = None |
| last_failure: Optional[datetime] = None |
| last_check: Optional[datetime] = None |
| avg_latency_ms: float = 0.0 |
| circuit_open: bool = False |
| circuit_opened_at: Optional[datetime] = None |
| metadata: dict = Field(default_factory=dict) |
|
|
|
|
| class SystemHealthSnapshot(BaseModel): |
| """Aggregate system health.""" |
| status: str = "healthy" |
| version: str = "1.0.0" |
| uptime_seconds: float = 0.0 |
| providers: List[ProviderHealthSnapshot] = Field(default_factory=list) |
| cache: dict = Field(default_factory=dict) |
| metrics: dict = Field(default_factory=dict) |
|
|
|
|
| class HealthSnapshot(BaseModel): |
| """Convenience alias.""" |
| system: SystemHealthSnapshot |
| timestamp: datetime = Field(default_factory=lambda: datetime.now(timezone.utc)) |
|
|