Honor Accept quality values when choosing the metrics format

prometheus_client/exposition.py

@@ -363,7 +363,21 @@ def choose_encoder(accept_header: str) -> Tuple[Callable[[Collector], bytes], st
     # Prometheus does.
     accept_header = accept_header or ''
     escaping = openmetrics.UNDERSCORES
+    accepted_types = []
     for accepted in accept_header.split(','):
+        quality = 1.0
+        for param in accepted.split(';')[1:]:
+            key, sep, value = param.strip().partition('=')
+            if sep and key.strip().lower() == 'q':
+                try:
+                    quality = float(value.strip())
+                except ValueError:
+                    quality = 0.0
+                break
+        if 0 < quality <= 1:
+            accepted_types.append((quality, accepted))
+
+    for _, accepted in sorted(accepted_types, key=lambda item: item[0], reverse=True):
         if accepted.split(';')[0].strip() == 'application/openmetrics-text':
             toks = accepted.split(';')
             version = _get_version(toks)
@@ -379,6 +393,8 @@ def choose_encoder(accept_header: str) -> Tuple[Callable[[Collector], bytes], st
             toks = accepted.split(';')
             version = _get_version(toks)
             escaping = _get_escaping(toks)
+            if not version or version == '0.0.4':
+                return generate_latest, CONTENT_TYPE_PLAIN_0_0_4
             # Only return an escaping header if we have a good version and
             # mimetype.
             if version and _version_at_least(version, (1, 0, 0)):

tests/test_exposition.py

@@ -648,6 +648,38 @@ def test_non_numeric_version(self):
                 self.assert_is_escaped(exp)
                 self.assert_is_prom(exp)
 
+    def test_accept_quality(self):
+        openmetrics_type = 'application/openmetrics-text; version=1.0.0'
+        text_type = 'text/plain; version=1.0.0'
+        for accept_header, expected_openmetrics in (
+            (f'{openmetrics_type};q=0, {text_type};q=1', False),
+            (f'{openmetrics_type};q=0.1, {text_type};q=0.9', False),
+            (f'{text_type};q=0.1, {openmetrics_type};q=0.9', True),
+            (f'{openmetrics_type};q=0.5, {text_type}', False),
+            (f'{text_type};q=0.5, {openmetrics_type}', True),
+            (f'{openmetrics_type};q=0.5, {text_type};q=0.5', True),
+            (f'{text_type};q=0.5, {openmetrics_type};q=0.5', False),
+            (f'{openmetrics_type};q=0.1, text/plain;version=0.0.4;q=0.9', False),
+            (f'{openmetrics_type};q=0.1, text/plain;q=0.9', False),
+        ):
+            with self.subTest(accept_header=accept_header):
+                generator, content_type = choose_encoder(accept_header)
+                self.assertEqual(content_type.startswith('application/openmetrics-text'), expected_openmetrics)
+                exp = generator(self.registry).decode('utf-8')
+                if expected_openmetrics:
+                    self.assert_is_openmetrics(exp)
+                else:
+                    self.assert_is_prom(exp)
+
+    def test_invalid_accept_quality(self):
+        for quality in ('bogus', '-1', '1.1', 'NaN', 'inf'):
+            with self.subTest(quality=quality):
+                generator, content_type = choose_encoder(
+                    f'application/openmetrics-text;version=1.0.0;q={quality}, text/plain;version=1.0.0'
+                )
+                self.assertTrue(content_type.startswith('text/plain'))
+                self.assert_is_prom(generator(self.registry).decode('utf-8'))
+
 
 class TestWsgiTLS(unittest.TestCase):
     def setUp(self):

tests/test_wsgi.py

@@ -66,6 +66,10 @@ def test_report_metrics_3(self):
     def test_report_metrics_4(self):
         self.validate_metrics("failed_requests", "Number of failed requests", 7)
 
+    def test_accept_quality(self):
+        self.environ['HTTP_ACCEPT'] = 'application/openmetrics-text;version=1.0.0;q=0, text/plain;version=0.0.4'
+        self.validate_metrics("counter", "A counter", 2)
+
     def test_favicon_path(self):
         from unittest.mock import patch