API Reference

This section provides detailed documentation for the aiopromql API.

Clients

class aiopromql.client.PrometheusAsync(url, timeout=2.0)[source]

Bases: PrometheusClientBase

Asynchronous Prometheus client using httpx.

async aclose()[source]

Close the async client session.

async query(promql, raw=False)[source]

Run an instant PromQL query.

Parameters:

  • promql (str) – The PromQL query string to execute.

  • raw (bool) – If True, return raw JSON response as dict; otherwise parse into model.

Return type:

Union[PrometheusResponseModel, dict]

Returns:

Parsed Prometheus response or raw JSON dict.

async query_range(promql, start, end, step='30s', raw=False)[source]

Run a ranged PromQL query over a time window.

Parameters:

  • promql (str) – The PromQL query string to execute.

  • start (datetime.datetime) – Start datetime of the query range.

  • end (datetime.datetime) – End datetime of the query range.

  • step (str) – Query resolution step width (e.g., ’30s’, ‘1m’).

  • raw (bool) – If True, return raw JSON response as dict; otherwise parse into model.

Return type:

Union[PrometheusResponseModel, dict]

Returns:

Parsed Prometheus response or raw JSON dict.

class aiopromql.client.PrometheusClientBase(url)[source]

Bases: object

Base Prometheus client with common utilities.

_parse_response(response)[source]

Parse Prometheus JSON response into model.

Return type:

PrometheusResponseModel

class aiopromql.client.PrometheusSync(url, timeout=2.0)[source]

Bases: PrometheusClientBase

Synchronous Prometheus client using httpx.

close()[source]

Close the sync client session.

query(promql, raw=False)[source]

Run an instant PromQL query.

Parameters:

  • promql (str) – The PromQL query string to execute.

  • raw (bool) – If True, return raw JSON response as dict; otherwise parse into model.

Return type:

Union[PrometheusResponseModel, dict]

Returns:

Parsed Prometheus response or raw JSON dict.

query_range(promql, start, end, step='30s', raw=False)[source]

Run a ranged PromQL query over a time window.

Parameters:

  • promql (str) – The PromQL query string to execute.

  • start (datetime.datetime) – Start datetime of the query range.

  • end (datetime.datetime) – End datetime of the query range.

  • step (str) – Query resolution step width (e.g., ’30s’, ‘1m’).

  • raw (bool) – If True, return raw JSON response as dict; otherwise parse into model.

Return type:

Union[PrometheusResponseModel, dict]

Returns:

Parsed Prometheus response or raw JSON dict.

Models

Core Models

Generic data structures for modeling time series and labeled metrics.

class aiopromql.models.core.MetricLabelSet(metric)[source]

Bases: object

Hashable wrapper around a Prometheus metric label dictionary.

Prometheus metrics are identified by a set of key-value labels (e.g., {“job”: “api”, “instance”: “localhost:9090”}). This class allows such a label set to be used as a key in Python dictionaries by making it hashable and comparable.

Instances of this class are used as keys in the dictionary returned by PrometheusResponseModel.to_metric_map(), where each MetricLabelSet maps to a TimeSeries object.

get(label, default=None)[source]

Return the value for the given label key, or default if not present.

Parameters:

  • label (str) – The label key to retrieve from the metric dictionary.

  • default – The value to return if the label is not found. Defaults to None.

Returns:

The value corresponding to the label, or the default if label is missing.

Return type:

str or Any

class aiopromql.models.core.TimeSeries(values)[source]

Bases: object

A sequence of timestamped float values (TimeSeriesPoint) with utility methods.

This class abstracts a Prometheus time series and provides methods for inspection, aggregation, and manipulation. Used in PrometheusResponseModel.to_metric_map() where each MetricLabelSet maps to a TimeSeries.

__init__(values)[source]
Parameters:

values (List[TimeSeriesPoint]) – List of initial TimeSeriesPoint objects.

add_point(point)[source]

Adds a new data point.

average()[source]

Computes the average of all values.

Return type:

float | None

extend(other)[source]

Appends another TimeSeries’ points to this one.

latest()[source]

Returns the latest (most recent) data point.

Return type:

TimeSeriesPoint | None

class aiopromql.models.core.TimeSeriesPoint(timestamp: datetime, value: float)[source]

Bases: NamedTuple

A single timestamped data point from a Prometheus time series.

Represents one (timestamp, value) pair, where the timestamp is a datetime object and the value is a float. Useful for building time series from Prometheus query results.

classmethod from_prometheus_value(ts, value)[source]

Converts a Prometheus response (timestamp, value) pair to TimeSeriesPoint.

Parameters:

  • ts (float) – Epoch timestamp.

  • value (str) – String representation of float value.

Return type:

TimeSeriesPoint

Returns:

A TimeSeriesPoint instance.

timestamp: datetime

Alias for field number 0

value: float

Alias for field number 1

Prometheus Models

Pydantic models for parsing and transforming Prometheus query json responses.

class aiopromql.models.prometheus.MatrixDataModel(**data)[source]

Bases: BaseModel

Parsed matrix data block from Prometheus.

model_config: ClassVar[ConfigDict] = {}

Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

result: List[MatrixResultModel]
resultType: Literal['matrix']
to_metric_map()[source]

Converts matrix results to a dict of TimeSeries grouped by metric labels.

Return type:

Dict[MetricLabelSet, TimeSeries]

Returns:

Dictionary mapping MetricLabelSet to TimeSeries.

class aiopromql.models.prometheus.MatrixResultModel(**data)[source]

Bases: BaseModel

Single Prometheus matrix result entry.

metric: Dict[str, str]
model_config: ClassVar[ConfigDict] = {}

Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

values: List[Tuple[float, str]]
class aiopromql.models.prometheus.PrometheusResponseModel(**data)[source]

Bases: BaseModel

Top-level Prometheus query response wrapper.

data: Union[VectorDataModel, MatrixDataModel]
model_config: ClassVar[ConfigDict] = {}

Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

status: Literal['success']
to_metric_map()[source]

Converts the response into a mapping from metric label sets to time series.

The resulting data structure uses generic, reusable types: - MetricLabelSet: a hashable representation of metric labels. - TimeSeries: a sequence of timestamped values.

Return type:

Dict[MetricLabelSet, TimeSeries]

Returns:

A dictionary mapping MetricLabelSet to TimeSeries.

class aiopromql.models.prometheus.VectorDataModel(**data)[source]

Bases: BaseModel

Parsed vector data block from Prometheus.

model_config: ClassVar[ConfigDict] = {}

Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

result: List[VectorResultModel]
resultType: Literal['vector']
to_metric_map()[source]

Converts vector results to a dict of TimeSeries object grouped by metric labels.

Return type:

Dict[MetricLabelSet, TimeSeries]

Returns:

Dictionary mapping MetricLabelSet to TimeSeries.

class aiopromql.models.prometheus.VectorResultModel(**data)[source]

Bases: BaseModel

Single Prometheus vector result entry.

metric: Dict[str, str]
model_config: ClassVar[ConfigDict] = {}

Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

value: Tuple[float, str]

Package

class aiopromql.PrometheusAsync(url, timeout=2.0)[source]

Bases: PrometheusClientBase

Asynchronous Prometheus client using httpx.

async aclose()[source]

Close the async client session.

async query(promql, raw=False)[source]

Run an instant PromQL query.

Parameters:

  • promql (str) – The PromQL query string to execute.

  • raw (bool) – If True, return raw JSON response as dict; otherwise parse into model.

Return type:

Union[PrometheusResponseModel, dict]

Returns:

Parsed Prometheus response or raw JSON dict.

async query_range(promql, start, end, step='30s', raw=False)[source]

Run a ranged PromQL query over a time window.

Parameters:

  • promql (str) – The PromQL query string to execute.

  • start (datetime.datetime) – Start datetime of the query range.

  • end (datetime.datetime) – End datetime of the query range.

  • step (str) – Query resolution step width (e.g., ’30s’, ‘1m’).

  • raw (bool) – If True, return raw JSON response as dict; otherwise parse into model.

Return type:

Union[PrometheusResponseModel, dict]

Returns:

Parsed Prometheus response or raw JSON dict.

class aiopromql.PrometheusSync(url, timeout=2.0)[source]

Bases: PrometheusClientBase

Synchronous Prometheus client using httpx.

close()[source]

Close the sync client session.

query(promql, raw=False)[source]

Run an instant PromQL query.

Parameters:

  • promql (str) – The PromQL query string to execute.

  • raw (bool) – If True, return raw JSON response as dict; otherwise parse into model.

Return type:

Union[PrometheusResponseModel, dict]

Returns:

Parsed Prometheus response or raw JSON dict.

query_range(promql, start, end, step='30s', raw=False)[source]

Run a ranged PromQL query over a time window.

Parameters:

  • promql (str) – The PromQL query string to execute.

  • start (datetime.datetime) – Start datetime of the query range.

  • end (datetime.datetime) – End datetime of the query range.

  • step (str) – Query resolution step width (e.g., ’30s’, ‘1m’).

  • raw (bool) – If True, return raw JSON response as dict; otherwise parse into model.

Return type:

Union[PrometheusResponseModel, dict]

Returns:

Parsed Prometheus response or raw JSON dict.

aiopromql.make_label_string(negate_keys=None, **labels)[source]

Return PromQL label selector string from provided labels.

negate_keys: iterable of keys whose match should be negated (using !=). labels: key=value pairs for labels.

Return type:

str