Created
January 19, 2022 15:18
-
-
Save petermeissner/b8706b2124af5849ee1d65d73f98f29f to your computer and use it in GitHub Desktop.
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
def dictkv_in_dictkv(dict_1: dict, dict_2: dict) -> bool: | |
""" | |
Function that checks if a set of key-values (expressed as dictionary) can be | |
found in another set of key-values (expressed as dictionary). | |
Args: | |
dict_1 (dict): find those kv (needles) | |
dict_2 (dict): search within these kv (haystack) | |
Returns: | |
bool: True if all can be found, False otherwise | |
""" | |
if len(dict_1) > len(dict_2): | |
return False | |
for i in dict_1.items(): | |
for j in dict_2.items(): | |
if dict_2.get(i[0], None) == i[1]: | |
pass | |
else: | |
return False | |
return True | |
class OpenMetrics: | |
""" | |
A Open Metrics aka Prometheus data format parser and handler. | |
Initialization will parse text into a list of items. | |
The filter method allows to filter metrics for name and/or labels. | |
""" | |
om_text = None | |
om_list = None | |
# initialize object via parsing string into sections | |
def __init__(self, text: str) -> None: | |
self.om_text = text | |
self.__to_list() | |
def __to_list(self) -> None: | |
# overwrite dict | |
self.om_list = list() | |
# go through sections/samples and make them a dict | |
for section in text_string_to_metric_families(self.om_text): | |
name = section.name | |
type = section.type | |
unit = section.unit | |
if len(section.samples) == 0: | |
self.om_list.append( | |
dict(name=name, type=type, unit=unit, labels=dict(), value=None) | |
) | |
else: | |
for s in section.samples: | |
self.om_list.append( | |
dict(name=name, type=type, unit=unit, labels=s.labels, value=s.value) | |
) | |
def filter(self, name=None, **kwargs): | |
# filter | |
res = [] | |
if name == None: | |
if len(kwargs) == 0: | |
res = self.om_list | |
else: | |
for item in self.om_list: | |
if dictkv_in_dictkv(dict_1=kwargs, dict_2=item['labels']): | |
res.append(item) | |
else: | |
for item in self.om_list: | |
labels_ok = dictkv_in_dictkv(dict_1=kwargs, dict_2=item['labels']) | |
name_ok = item['name'] == name | |
if labels_ok & name_ok : | |
res.append(item) | |
# return | |
return res | |
Sign up for free
to join this conversation on GitHub.
Already have an account?
Sign in to comment