Simple library I used for querying Prometheus API and storing the results in Polars dataframes

prom.py

import os

from datetime import datetime, timedelta
from requests import get
from typing import Tuple


# https://prometheus.io/docs/prometheus/latest/querying/api/


class Prometheus:
    """
    Class for querying Prometheus
    """

    def __init__(
        self,
        url: str = os.getenv("PROMETHEUS_URL"),
        auth: Tuple = (os.getenv("PROMETHEUS_USER"), os.getenv("PROMETHEUS_PASS")),
    ) -> None:
        self.url = url
        self.auth = auth

    def labels(
        self,
        start: datetime = datetime.now() - timedelta(days=1),
        end: datetime = datetime.now(),
    ) -> list:
        """
        List all labels
        """

        params = (
            {}
            if start == "" or end == ""
            else {
                "start": start.astimezone().isoformat(),
                "end": end.astimezone().isoformat(),
            }
        )

        res = get(url=self.url + "/api/v1/labels", auth=self.auth, params=params)

        if res.status_code != 200 or res.json()["status"] != "success":
            raise Exception("Well, fuck...")

        return res.json()["data"]

    def values(
        self,
        label: str,
        start: datetime = datetime.now() - timedelta(days=1),
        end: datetime = datetime.now(),
    ) -> list:
        """
        List all values of given label
        """

        params = (
            {}
            if start == "" or end == ""
            else {
                "start": start.astimezone().isoformat(),
                "end": end.astimezone().isoformat(),
            }
        )

        res = get(
            url=self.url + "/api/v1/label/" + label + "/values",
            auth=self.auth,
            params=params,
        )

        if res.status_code != 200 or res.json()["status"] != "success":
            raise Exception("Well, fuck...")

        return res.json()["data"]

    def query(self, expression: str = '{job=~".+"}', time: datetime = datetime.now()) -> Tuple[str, list]:
        """
        An instant query expression
        """

        res = get(
            url=self.url + "/api/v1/query",
            auth=self.auth,
            params={
                "query": expression,
                "time": time.astimezone().isoformat(),
            },
        )

        if res.status_code != 200 or res.json()["status"] != "success":
            # print(res.text)
            raise Exception("Well, fuck...")

        return res.json()["data"]["resultType"], res.json()["data"]["result"]

    def queryRange(
        self,
        expression: str = '{job=~".+"}',
        start: datetime = datetime.now() - timedelta(days=1),
        end: datetime = datetime.now(),
    ) -> Tuple[str, list]:
        """
        A range query expression
        """

        res = get(
            url=self.url + "/api/v1/query_range",
            auth=self.auth,
            params={
                "query": expression,
                "start": start.astimezone().isoformat(),
                "end": end.astimezone().isoformat(),
                "step": int((end - start).total_seconds()) // 10000,  # this is hardcoded limit
            },
        )

        if res.status_code != 200 or res.json()["status"] != "success":
            print(res.text)
            raise Exception("Well, fuck...")

        return res.json()["data"]["resultType"], res.json()["data"]["result"]

    def queryName(self, name: str) -> Tuple[str, list]:
        """
        Instant, return raw metrics from type "vector", based on provided __name__ label
        """

        res = self.query(expression='{__name__="%s"}' % name)

        if res[0] != "vector":
            raise Exception("Well, fuck...")

        return res[1]

    def queryRangeName(
        self,
        name: str,
        start: datetime = datetime.now() - timedelta(days=1),
        end: datetime = datetime.now(),
    ) -> list:
        """
        Range, return raw metrics from type "matrix", based on provided __name__ label
        """

        res = self.queryRange(
            expression='{__name__="%s"}' % name,
            start=start.astimezone().isoformat(),
            end=end.astimezone().isoformat(),
        )

        if res[0] != "matrix":
            raise Exception("Well, fuck...")

        return res[1]