(labels_string: str, openmetrics: bool = False)
| 50 | |
| 51 | |
| 52 | def parse_labels(labels_string: str, openmetrics: bool = False) -> Dict[str, str]: |
| 53 | labels: Dict[str, str] = {} |
| 54 | |
| 55 | # Copy original labels |
| 56 | sub_labels = labels_string.strip() |
| 57 | if openmetrics and sub_labels and sub_labels[0] == ',': |
| 58 | raise ValueError("leading comma: " + labels_string) |
| 59 | try: |
| 60 | # Process one label at a time |
| 61 | while sub_labels: |
| 62 | # The label name is before the equal, or if there's no equal, that's the |
| 63 | # metric name. |
| 64 | |
| 65 | name_term, value_term, sub_labels = _next_term(sub_labels, openmetrics) |
| 66 | if not value_term: |
| 67 | if openmetrics: |
| 68 | raise ValueError("empty term in line: " + labels_string) |
| 69 | continue |
| 70 | |
| 71 | label_name, quoted_name = _unquote_unescape(name_term) |
| 72 | |
| 73 | if not quoted_name and not _is_valid_legacy_metric_name(label_name): |
| 74 | raise ValueError("unquoted UTF-8 metric name") |
| 75 | |
| 76 | # Check for missing quotes |
| 77 | if not value_term or value_term[0] != '"': |
| 78 | raise ValueError |
| 79 | |
| 80 | # The first quote is guaranteed to be after the equal. |
| 81 | # Make sure that the next unescaped quote is the last character. |
| 82 | i = 1 |
| 83 | while i < len(value_term): |
| 84 | i = value_term.index('"', i) |
| 85 | if not _is_character_escaped(value_term[:i], i): |
| 86 | break |
| 87 | i += 1 |
| 88 | # The label value is between the first and last quote |
| 89 | quote_end = i + 1 |
| 90 | if quote_end != len(value_term): |
| 91 | raise ValueError("unexpected text after quote: " + labels_string) |
| 92 | |
| 93 | label_value, _ = _unquote_unescape(value_term) |
| 94 | if label_name == '__name__': |
| 95 | _validate_metric_name(label_name) |
| 96 | else: |
| 97 | _validate_labelname(label_name) |
| 98 | if label_name in labels: |
| 99 | raise ValueError("invalid line, duplicate label name: " + labels_string) |
| 100 | labels[label_name] = label_value |
| 101 | return labels |
| 102 | except ValueError: |
| 103 | raise ValueError("Invalid labels: " + labels_string) |
| 104 | |
| 105 | |
| 106 | def _next_term(text: str, openmetrics: bool) -> Tuple[str, str, str]: |
no test coverage detected