Skip to content

Instantly share code, notes, and snippets.

@jediminer543
Created May 31, 2022
Embed
What would you like to do?
'''
Bar2Prom
Brought to you by JMT Makes Thinks
CC0
Written by jediminer543
'''
import requests
from prometheus_client.core import GaugeMetricFamily, CounterMetricFamily, REGISTRY
from prometheus_client import start_http_server
endpoint = "https://bar.emf.camp/api/stocktypes.json"
class BarCollector():
value = 0
def collect(self):
response = requests.get(endpoint)
stocks = response.json()
stocks = stocks["stocktypes"]
base_units = []
for stock in stocks:
bu = stock["base_unit"]
if bu not in base_units:
base_units.append(bu)
base_units_bought = {}
base_units_remaining = {}
for unit in base_units:
base_units_bought[unit] = CounterMetricFamily("base_units_bought", "the amount of stock bought in base units as a Decimal", labels=["id", "department_id", "manufacturer", "name"], unit=unit)
base_units_remaining[unit] = GaugeMetricFamily("base_units_bought", "the amount of stock bought in base units as a Decimal", labels=["id", "department_id", "manufacturer", "name"], unit=unit)
for stock in stocks:
bu = stock["base_unit"]
base_units_bought[bu].add_metric([str(stock["id"]), str(stock["department"]["id"]), str(stock["manufacturer"]), str(stock["name"])], float(stock["base_units_bought"]))
base_units_remaining[bu].add_metric([str(stock["id"]), str(stock["department"]["id"]), str(stock["manufacturer"]), str(stock["name"])], float(stock["base_units_remaining"]))
for bub in base_units_bought.items():
yield bub[1]
for bur in base_units_remaining.items():
yield bur[1]
yield CounterMetricFamily("run_count", "Number of fetched times", value=self.value)
self.value += 1
if __name__ == "__main__":
REGISTRY.register(BarCollector())
start_http_server(8000)
while True:
pass
Sign up for free to join this conversation on GitHub. Already have an account? Sign in to comment