"""Phase 13: Operational Alerts — Pydantic schemas.""" from __future__ import annotations from datetime import datetime from typing import Any, Dict, List, Optional from uuid import UUID from pydantic import BaseModel, Field, field_validator from app.models.operational_alert import AlertRuleType, AlertSeverity, AlertStatus VALID_SEVERITIES = {s.value for s in AlertSeverity} VALID_STATUSES = {s.value for s in AlertStatus} VALID_RULE_TYPES = {r.value for r in AlertRuleType} # ── AlertRule schemas ───────────────────────────────────────────────────────── class AlertRuleCreate(BaseModel): name: str = Field(..., min_length=1, max_length=300) description: Optional[str] = None rule_type: str severity: str = "medium" config: Dict[str, Any] = Field(default_factory=dict) notify_in_app: bool = True notify_webhook: bool = False webhook_id: Optional[UUID] = None cooldown_hours: int = Field(24, ge=0, le=8760) @field_validator("rule_type") @classmethod def validate_rule_type(cls, v: str) -> str: if v not in VALID_RULE_TYPES: raise ValueError(f"Invalid rule_type. Valid: {VALID_RULE_TYPES}") return v @field_validator("severity") @classmethod def validate_severity(cls, v: str) -> str: if v not in VALID_SEVERITIES: raise ValueError(f"Invalid severity. Valid: {VALID_SEVERITIES}") return v class AlertRuleUpdate(BaseModel): name: Optional[str] = Field(None, min_length=1, max_length=300) description: Optional[str] = None severity: Optional[str] = None is_enabled: Optional[bool] = None config: Optional[Dict[str, Any]] = None notify_in_app: Optional[bool] = None notify_webhook: Optional[bool] = None webhook_id: Optional[UUID] = None cooldown_hours: Optional[int] = Field(None, ge=0, le=8760) @field_validator("severity") @classmethod def validate_severity(cls, v: Optional[str]) -> Optional[str]: if v is not None and v not in VALID_SEVERITIES: raise ValueError(f"Invalid severity. Valid: {VALID_SEVERITIES}") return v class AlertRuleOut(BaseModel): id: UUID name: str description: Optional[str] = None rule_type: str severity: str is_enabled: bool is_system: bool config: Dict[str, Any] notify_in_app: bool notify_webhook: bool webhook_id: Optional[UUID] = None cooldown_hours: int created_by: Optional[UUID] = None created_at: Optional[datetime] = None last_fired_at: Optional[datetime] = None class Config: from_attributes = True # ── AlertInstance schemas ───────────────────────────────────────────────────── class AlertInstanceOut(BaseModel): id: UUID rule_id: Optional[UUID] = None rule_name: str rule_type: str severity: str title: str message: str details: Optional[Dict[str, Any]] = None status: str acknowledged_by: Optional[UUID] = None acknowledged_at: Optional[datetime] = None resolved_at: Optional[datetime] = None created_at: Optional[datetime] = None class Config: from_attributes = True # ── Evaluation result ───────────────────────────────────────────────────────── class EvaluationResult(BaseModel): rules_evaluated: int alerts_fired: int alerts: List[AlertInstanceOut] = Field(default_factory=list) duration_seconds: float # ── Summary ─────────────────────────────────────────────────────────────────── class AlertSummary(BaseModel): total_open: int total_acknowledged: int total_resolved: int by_severity: Dict[str, int] by_rule_type: Dict[str, int] recent_alerts: List[AlertInstanceOut] = Field(default_factory=list)