""" Pydantic models for Metric Discovery (list_metrics) tool. Optional match parameter is passed through to Prometheus as match[] series selector. """ from datetime import datetime from typing import Optional from pydantic import ( # type: ignore # pydantic in pyproject; IDE may use different interpreter BaseModel, Field, field_validator, ) MAX_LIST_METRICS_LIMIT = 500 class ListMetricsRequest(BaseModel): """ Input model for list_metrics requests. Optional match (string): regex or prefix to filter metric names. """ match: Optional[str] = Field( default=None, description=( "Optional Prometheus series selector used as match[] " "(e.g. '{__name__=~\"topsql_.*\"}')." ), ) cluster_id: str = Field( ..., description="Cluster identifier", ) env: str = Field( ..., description="Environment name used in biz_type resolution (e.g. prod, staging, dev)", ) deploy_type: str = Field( ..., description=( "Cluster deploy type used in biz_type resolution " "(dedicated, premium, resource_pool, starter, essential)" ), ) limit: int = Field( default=MAX_LIST_METRICS_LIMIT, ge=1, le=MAX_LIST_METRICS_LIMIT, description=(f"Maximum number of metric names to return (1-{MAX_LIST_METRICS_LIMIT})."), ) start: Optional[str] = Field( default=None, description="Optional start timestamp in RFC3339 format.", ) end: Optional[str] = Field( default=None, description="Optional end timestamp in RFC3339 format.", ) @field_validator("match") @classmethod def normalize_match(cls, v: Optional[str]) -> Optional[str]: if v is None or not v.strip(): return None return v.strip() @field_validator("start", "end") @classmethod def validate_timestamp(cls, v: Optional[str]) -> Optional[str]: if v is None or not v.strip(): return None value = v.strip() try: datetime.fromisoformat(value.replace("Z", "+00:00")) return value except ValueError as e: raise ValueError(f"Invalid timestamp format: {value}. Must be RFC3339.") from e @field_validator("env", "deploy_type", mode="before") @classmethod def normalize_scope_text(cls, v: str) -> str: return v.strip().lower() @field_validator("cluster_id", mode="before") @classmethod def normalize_cluster_id(cls, v: object) -> str: if not isinstance(v, str): raise ValueError("cluster_id must be a string. Quote numeric IDs explicitly.") return str(v).strip()