"""
Pydantic models for PromQL query responses.
These models provide type-safe representations of Prometheus API responses.
"""
from typing import Any, Literal
from pydantic import BaseModel, Field
class Sample(BaseModel):
"""A single sample (timestamp, value) from an instant query."""
timestamp: float = Field(description="Unix timestamp")
value: str = Field(description="Sample value as string")
@classmethod
def from_prometheus(cls, data: list[Any]) -> "Sample":
"""Create a Sample from Prometheus [timestamp, value] format."""
return cls(timestamp=data[0], value=data[1])
class TimeSeries(BaseModel):
"""A time series with metric labels and samples."""
metric: dict[str, str] = Field(
default_factory=dict,
description="Metric labels",
)
values: list[Sample] = Field(
default_factory=list,
description="Sample values (for range queries)",
)
value: Sample | None = Field(
default=None,
description="Single sample value (for instant queries)",
)
@classmethod
def from_instant_result(cls, data: dict[str, Any]) -> "TimeSeries":
"""Create a TimeSeries from an instant query result."""
metric = data.get("metric", {})
value_data = data.get("value", [0, "0"])
return cls(
metric=metric,
value=Sample.from_prometheus(value_data),
)
@classmethod
def from_range_result(cls, data: dict[str, Any]) -> "TimeSeries":
"""Create a TimeSeries from a range query result."""
metric = data.get("metric", {})
values = [Sample.from_prometheus(v) for v in data.get("values", [])]
return cls(metric=metric, values=values)
class QueryResult(BaseModel):
"""Result of an instant PromQL query."""
status: Literal["success", "error"] = Field(description="Query status")
result_type: Literal["vector", "scalar", "string", "matrix"] | None = Field(
default=None,
description="Type of query result",
)
result: list[TimeSeries] = Field(
default_factory=list,
description="Query result data",
)
error: str | None = Field(default=None, description="Error message if status is error")
error_type: str | None = Field(default=None, description="Error type if status is error")
@classmethod
def from_prometheus_response(cls, response: dict[str, Any]) -> "QueryResult":
"""Create a QueryResult from a Prometheus API response."""
status = response.get("status", "error")
data = response.get("data", {})
if status == "error":
return cls(
status="error",
error=response.get("error", "Unknown error"),
error_type=response.get("errorType"),
)
result_type = data.get("resultType")
raw_result = data.get("result", [])
if result_type == "vector":
result = [TimeSeries.from_instant_result(r) for r in raw_result]
elif result_type == "scalar":
result = [
TimeSeries(
metric={},
value=Sample.from_prometheus(raw_result),
)
]
else:
result = []
return cls(
status="success",
result_type=result_type,
result=result,
)
class RangeQueryResult(BaseModel):
"""Result of a range PromQL query."""
status: Literal["success", "error"] = Field(description="Query status")
result_type: Literal["matrix"] | None = Field(
default=None,
description="Type of query result (always 'matrix' for range queries)",
)
result: list[TimeSeries] = Field(
default_factory=list,
description="Query result data as time series",
)
error: str | None = Field(default=None, description="Error message if status is error")
error_type: str | None = Field(default=None, description="Error type if status is error")
@classmethod
def from_prometheus_response(cls, response: dict[str, Any]) -> "RangeQueryResult":
"""Create a RangeQueryResult from a Prometheus API response."""
status = response.get("status", "error")
data = response.get("data", {})
if status == "error":
return cls(
status="error",
error=response.get("error", "Unknown error"),
error_type=response.get("errorType"),
)
result_type = data.get("resultType")
raw_result = data.get("result", [])
result = [TimeSeries.from_range_result(r) for r in raw_result]
return cls(
status="success",
result_type=result_type,
result=result,
)
class LabelList(BaseModel):
"""List of label names from Prometheus."""
status: Literal["success", "error"] = Field(description="Query status")
labels: list[str] = Field(default_factory=list, description="List of label names")
error: str | None = Field(default=None, description="Error message if status is error")
@classmethod
def from_prometheus_response(cls, response: dict[str, Any]) -> "LabelList":
"""Create a LabelList from a Prometheus API response."""
status = response.get("status", "error")
if status == "error":
return cls(
status="error",
error=response.get("error", "Unknown error"),
)
return cls(
status="success",
labels=response.get("data", []),
)
class LabelValues(BaseModel):
"""List of values for a specific label."""
status: Literal["success", "error"] = Field(description="Query status")
label_name: str = Field(description="The label name these values belong to")
values: list[str] = Field(default_factory=list, description="List of label values")
error: str | None = Field(default=None, description="Error message if status is error")
@classmethod
def from_prometheus_response(
cls,
response: dict[str, Any],
label_name: str,
) -> "LabelValues":
"""Create a LabelValues from a Prometheus API response."""
status = response.get("status", "error")
if status == "error":
return cls(
status="error",
label_name=label_name,
error=response.get("error", "Unknown error"),
)
return cls(
status="success",
label_name=label_name,
values=response.get("data", []),
)
class MetricInfo(BaseModel):
"""Information about a metric including metadata."""
name: str = Field(description="Metric name")
type: str | None = Field(default=None, description="Metric type (counter, gauge, histogram, summary)")
help: str | None = Field(default=None, description="Metric description/help text")
unit: str | None = Field(default=None, description="Metric unit")
class MetricList(BaseModel):
"""List of metrics with optional metadata."""
status: Literal["success", "error"] = Field(description="Query status")
metrics: list[MetricInfo] = Field(
default_factory=list,
description="List of metrics",
)
error: str | None = Field(default=None, description="Error message if status is error")
@classmethod
def from_label_values_response(cls, response: dict[str, Any]) -> "MetricList":
"""Create a MetricList from a label values response for __name__."""
status = response.get("status", "error")
if status == "error":
return cls(
status="error",
error=response.get("error", "Unknown error"),
)
metric_names = response.get("data", [])
metrics = [MetricInfo(name=name) for name in metric_names]
return cls(status="success", metrics=metrics)
@classmethod
def from_metadata_response(cls, response: dict[str, Any]) -> "MetricList":
"""Create a MetricList from a metadata response with full info."""
status = response.get("status", "error")
if status == "error":
return cls(
status="error",
error=response.get("error", "Unknown error"),
)
metadata = response.get("data", {})
metrics = []
for name, info_list in metadata.items():
# Take the first metadata entry for each metric
info = info_list[0] if info_list else {}
metrics.append(
MetricInfo(
name=name,
type=info.get("type"),
help=info.get("help"),
unit=info.get("unit"),
)
)
return cls(status="success", metrics=metrics)
