Add typing to Histogram/Summary/Gauge

mypy.ini

@@ -1,6 +1,7 @@
 [mypy]
 exclude = prometheus_client/decorator.py|prometheus_client/twisted|tests/test_twisted.py
 implicit_reexport = False
+disallow_incomplete_defs = True
 
 [mypy-prometheus_client.decorator]
 follow_imports = skip

prometheus_client/metrics.py

@@ -2,7 +2,7 @@
 import time
 import types
 from typing import (
-    Any, Callable, Dict, Iterable, Optional, Sequence, Type, TypeVar,
+    Any, Callable, Dict, Iterable, Optional, Sequence, Type, TypeVar, Union,
 )
 
 from . import values  # retain this import style for testability
@@ -337,15 +337,15 @@ def f():
     _MULTIPROC_MODES = frozenset(('min', 'max', 'livesum', 'liveall', 'all'))
 
     def __init__(self,
-                 name,
-                 documentation,
-                 labelnames=(),
-                 namespace='',
-                 subsystem='',
-                 unit='',
-                 registry=REGISTRY,
-                 _labelvalues=None,
-                 multiprocess_mode='all',
+                 name: str,
+                 documentation: str,
+                 labelnames: Iterable[str] = (),
+                 namespace: str = '',
+                 subsystem: str = '',
+                 unit: str = '',
+                 registry: Optional[CollectorRegistry] = REGISTRY,
+                 _labelvalues: Optional[Sequence[str]] = None,
+                 multiprocess_mode: str = 'all',
                  ):
         self._multiprocess_mode = multiprocess_mode
         if multiprocess_mode not in self._MULTIPROC_MODES:
@@ -362,32 +362,32 @@ def __init__(self,
         )
         self._kwargs['multiprocess_mode'] = self._multiprocess_mode
 
-    def _metric_init(self):
+    def _metric_init(self) -> None:
         self._value = values.ValueClass(
             self._type, self._name, self._name, self._labelnames, self._labelvalues,
             multiprocess_mode=self._multiprocess_mode
         )
 
-    def inc(self, amount=1):
+    def inc(self, amount: float = 1) -> None:
         """Increment gauge by the given amount."""
         self._raise_if_not_observable()
         self._value.inc(amount)
 
-    def dec(self, amount=1):
+    def dec(self, amount: float = 1) -> None:
         """Decrement gauge by the given amount."""
         self._raise_if_not_observable()
         self._value.inc(-amount)
 
-    def set(self, value):
+    def set(self, value: float) -> None:
         """Set gauge to the given value."""
         self._raise_if_not_observable()
         self._value.set(float(value))
 
-    def set_to_current_time(self):
+    def set_to_current_time(self) -> None:
         """Set gauge to the current unixtime."""
         self.set(time.time())
 
-    def track_inprogress(self):
+    def track_inprogress(self) -> InprogressTracker:
         """Track inprogress blocks of code or functions.
 
         Can be used as a function decorator or context manager.
@@ -397,14 +397,14 @@ def track_inprogress(self):
         self._raise_if_not_observable()
         return InprogressTracker(self)
 
-    def time(self):
+    def time(self) -> Timer:
         """Time a block of code or function, and set the duration in seconds.
 
         Can be used as a function decorator or context manager.
         """
         return Timer(self, 'set')
 
-    def set_function(self, f):
+    def set_function(self, f: Callable[[], float]) -> None:
         """Call the provided function to return the Gauge value.
 
         The function must return a float, and may be called from
@@ -413,10 +413,10 @@ def set_function(self, f):
 
         self._raise_if_not_observable()
 
-        def samples(self) -> Iterable[Sample]:
+        def samples(_: Gauge) -> Iterable[Sample]:
             return (Sample('', {}, float(f()), None, None),)
 
-        self._child_samples = types.MethodType(samples, self)
+        self._child_samples = types.MethodType(samples, self)  # type: ignore
 
     def _child_samples(self) -> Iterable[Sample]:
         return (Sample('', {}, self._value.get(), None, None),)
@@ -455,13 +455,13 @@ def create_response(request):
     _type = 'summary'
     _reserved_labelnames = ['quantile']
 
-    def _metric_init(self):
+    def _metric_init(self) -> None:
         self._count = values.ValueClass(self._type, self._name, self._name + '_count', self._labelnames,
                                         self._labelvalues)
         self._sum = values.ValueClass(self._type, self._name, self._name + '_sum', self._labelnames, self._labelvalues)
         self._created = time.time()
 
-    def observe(self, amount):
+    def observe(self, amount: float) -> None:
         """Observe the given amount.
 
         The amount is usually positive or zero. Negative values are
@@ -475,7 +475,7 @@ def observe(self, amount):
         self._count.inc(1)
         self._sum.inc(amount)
 
-    def time(self):
+    def time(self) -> Timer:
         """Time a block of code or function, and observe the duration in seconds.
 
         Can be used as a function decorator or context manager.
@@ -530,15 +530,15 @@ def create_response(request):
     DEFAULT_BUCKETS = (.005, .01, .025, .05, .075, .1, .25, .5, .75, 1.0, 2.5, 5.0, 7.5, 10.0, INF)
 
     def __init__(self,
-                 name,
-                 documentation,
-                 labelnames=(),
-                 namespace='',
-                 subsystem='',
-                 unit='',
-                 registry=REGISTRY,
-                 _labelvalues=None,
-                 buckets=DEFAULT_BUCKETS,
+                 name: str,
+                 documentation: str,
+                 labelnames: Iterable[str] = (),
+                 namespace: str = '',
+                 subsystem: str = '',
+                 unit: str = '',
+                 registry: Optional[CollectorRegistry] = REGISTRY,
+                 _labelvalues: Optional[Sequence[str]] = None,
+                 buckets: Sequence[Union[float, int, str]] = DEFAULT_BUCKETS,
                  ):
         self._prepare_buckets(buckets)
         super().__init__(
@@ -553,8 +553,8 @@ def __init__(self,
         )
         self._kwargs['buckets'] = buckets
 
-    def _prepare_buckets(self, buckets):
-        buckets = [float(b) for b in buckets]
+    def _prepare_buckets(self, source_buckets: Sequence[Union[float, int, str]]) -> None:
+        buckets = [float(b) for b in source_buckets]
         if buckets != sorted(buckets):
             # This is probably an error on the part of the user,
             # so raise rather than sorting for them.
@@ -565,7 +565,7 @@ def _prepare_buckets(self, buckets):
             raise ValueError('Must have at least two buckets')
         self._upper_bounds = buckets
 
-    def _metric_init(self):
+    def _metric_init(self) -> None:
         self._buckets = []
         self._created = time.time()
         bucket_labelnames = self._labelnames + ('le',)
@@ -579,7 +579,7 @@ def _metric_init(self):
                 self._labelvalues + (floatToGoString(b),))
             )
 
-    def observe(self, amount, exemplar=None):
+    def observe(self, amount: float, exemplar: Optional[Dict[str, str]] = None) -> None:
         """Observe the given amount.
 
         The amount is usually positive or zero. Negative values are
@@ -599,7 +599,7 @@ def observe(self, amount, exemplar=None):
                     self._buckets[i].set_exemplar(Exemplar(exemplar, amount, time.time()))
                 break
 
-    def time(self):
+    def time(self) -> Timer:
         """Time a block of code or function, and observe the duration in seconds.
 
         Can be used as a function decorator or context manager.