A simple use of python prometheus lib with PushGateWay

prometheus_pushgateway_ready_to_use.py

"""
# About
Essentially, in an app like Django, you can not simply make 
'from prometheus_client import push_to_gateway' and go on as you may expect. 
You will face with several issues, e.g. use only one CollectorRegistery 
and basically, you don't want to deal with the ones when importing push_2_gateway. 

So, the classes below give you a possibility (python overlay) you looking for: 
import metric class and simply push a value to the metric.

# Agenda
- builder class that consists of CollectorRegistery and metrics already registered in a process
- general (not all) prometheuse metric classes that in ready-to-use state

# The Usage:
Let me show the usage for PGInfo (i.e. prometheuse INFO metric)
>>> from prometheus_pushgateway_ready_to_use import PGInfo
>>> prometheus_info_metric = PGInfo(name='my_name', description='my description')
>>> prometheus_info_metric.set({'quote': 'the labour we love physics pain'})

## Where to find the values?
In appropriate Prometheus you can seek the metric pushed to pushgateway by the following format:
>>> my_name_info{job="default", quote="the labour we love physics pain"}  # Take a note on tralling '_info'
"""
from typing import Union
from prometheus_client import (
    CollectorRegistry, Gauge, push_to_gateway, delete_from_gateway, Counter, Info
)

PUSHGATEWAY_API = 'pushgateway:9091'


class PGMetricBuilder:
    _registery = CollectorRegistry()
    _metrics_registered = dict()

    @classmethod
    def _not_defined_metric_class(cls, **kwargs):
        raise NotImplemented

    # There are more metrics in prometheus_client that you can pick...
    metric_class: Union[Counter, Gauge, Info] = _not_defined_metric_class

    def __init__(self, name: str, description: str = "No description", job: str = 'default'):
        """
        Get or create a metric via the builder with allocation to
        PushGateWay instance. When inherit class define metric_class property.

        :param name: name of ur metric. It is a uniq key
        :param description: optionally description
        :param job: optionally, https://prometheus.io/docs/concepts/jobs_instances/
        """
        if name in PGMetricBuilder._metrics_registered:
            self.metric = PGMetricBuilder._metrics_registered.get(name)
        else:
            self.metric = self.metric_class(name=name, documentation=description, registry=PGMetricBuilder._registery)
            PGMetricBuilder._metrics_registered[name] = self.metric
        self.job = job

    def _push_metric(self):
        """ After each of metric processing dnf to push delta.

        Note: dnf that prometheus pulls PGW with a constant period
        that might be greater than the period you update metric in PGW.
        """
        push_to_gateway(PUSHGATEWAY_API, job=self.job, registry=PGMetricBuilder._registery)

    def delete(self):
        delete_from_gateway(PUSHGATEWAY_API, job=self.job)


class PGCounter(PGMetricBuilder):
    """ Discoverable metrics: <name>{_created, _total}"""
    metric_class = Counter

    def __init__(self, *args, **kwargs):
        super().__init__(*args, **kwargs)

    def inc(self, value: float = 1):
        self.metric.inc(value)
        self._push_metric()


class PGInfo(PGMetricBuilder):
    """ Discoverable metrics: <name>{_info} """
    metric_class = Info

    def __init__(self, *args, **kwargs):
        super().__init__(*args, **kwargs)

    def set(self, dict2save: dict):
        self.metric.info(dict2save)
        self._push_metric()


class PGGauge(PGMetricBuilder):
    """ Discoverable metrics: <name> """
    metric_class = Gauge

    def __init__(self, *args, **kwargs):
        super().__init__(*args, **kwargs)

    def set(self, value: int = 0):
        self.metric.set(value)
        self._push_metric()

    def inc(self, value: float = 1):
        self.metric.inc(value)
        self._push_metric()

    def dec(self, value: float):
        self.metric.dec(value)
        self._push_metric()