1
1
import io as StringIO
2
2
import re
3
+ from typing import Dict , Iterable , List , Match , Optional , TextIO , Tuple
3
4
4
5
from .metrics_core import Metric
5
6
from .samples import Sample
6
7
7
8
8
- def text_string_to_metric_families (text ) :
9
+ def text_string_to_metric_families (text : str ) -> Iterable [ Metric ] :
9
10
"""Parse Prometheus text format from a unicode string.
10
11
11
12
See text_fd_to_metric_families.
@@ -20,32 +21,32 @@ def text_string_to_metric_families(text):
20
21
}
21
22
22
23
23
- def replace_escape_sequence (match ) :
24
+ def replace_escape_sequence (match : Match [ str ]) -> str :
24
25
return ESCAPE_SEQUENCES [match .group (0 )]
25
26
26
27
27
28
HELP_ESCAPING_RE = re .compile (r'\\[\\n]' )
28
29
ESCAPING_RE = re .compile (r'\\[\\n"]' )
29
30
30
31
31
- def _replace_help_escaping (s ) :
32
+ def _replace_help_escaping (s : str ) -> str :
32
33
return HELP_ESCAPING_RE .sub (replace_escape_sequence , s )
33
34
34
35
35
- def _replace_escaping (s ) :
36
+ def _replace_escaping (s : str ) -> str :
36
37
return ESCAPING_RE .sub (replace_escape_sequence , s )
37
38
38
39
39
- def _is_character_escaped (s , charpos ) :
40
+ def _is_character_escaped (s : str , charpos : int ) -> bool :
40
41
num_bslashes = 0
41
42
while (charpos > num_bslashes
42
43
and s [charpos - 1 - num_bslashes ] == '\\ ' ):
43
44
num_bslashes += 1
44
45
return num_bslashes % 2 == 1
45
46
46
47
47
- def _parse_labels (labels_string ) :
48
- labels = {}
48
+ def _parse_labels (labels_string : str ) -> Dict [ str , str ] :
49
+ labels : Dict [ str , str ] = {}
49
50
# Return if we don't have valid labels
50
51
if "=" not in labels_string :
51
52
return labels
@@ -95,7 +96,7 @@ def _parse_labels(labels_string):
95
96
96
97
97
98
# If we have multiple values only consider the first
98
- def _parse_value_and_timestamp (s ) :
99
+ def _parse_value_and_timestamp (s : str ) -> Tuple [ float , Optional [ float ]] :
99
100
s = s .lstrip ()
100
101
separator = " "
101
102
if separator not in s :
@@ -108,7 +109,7 @@ def _parse_value_and_timestamp(s):
108
109
return value , timestamp
109
110
110
111
111
- def _parse_sample (text ) :
112
+ def _parse_sample (text : str ) -> Sample :
112
113
# Detect the labels in the text
113
114
try :
114
115
label_start , label_end = text .index ("{" ), text .rindex ("}" )
@@ -133,7 +134,7 @@ def _parse_sample(text):
133
134
return Sample (name , {}, value , timestamp )
134
135
135
136
136
- def text_fd_to_metric_families (fd ) :
137
+ def text_fd_to_metric_families (fd : TextIO ) -> Iterable [ Metric ] :
137
138
"""Parse Prometheus text format from a file descriptor.
138
139
139
140
This is a laxer parser than the main Go parser,
@@ -145,10 +146,10 @@ def text_fd_to_metric_families(fd):
145
146
name = ''
146
147
documentation = ''
147
148
typ = 'untyped'
148
- samples = []
149
+ samples : List [ Sample ] = []
149
150
allowed_names = []
150
151
151
- def build_metric (name , documentation , typ , samples ) :
152
+ def build_metric (name : str , documentation : str , typ : str , samples : List [ Sample ]) -> Metric :
152
153
# Munge counters into OpenMetrics representation
153
154
# used internally.
154
155
if typ == 'counter' :
0 commit comments