fix typing to support py 3.9

numaprom/monitoring/metrics.py

@@ -1,4 +1,5 @@
 import logging
+from typing import Optional
 
 from prometheus_client import Counter, Info, Summary, Gauge, Histogram
 
@@ -21,22 +22,22 @@ def __init__(
         self,
         name: str,
         description: str,
-        label_pairs: dict[str, str] | None,
+        label_pairs: dict,
     ) -> None:
         self.label_pairs = dict(label_pairs)  # converting DictConfig to dict type
         self.name = name
         self.description = description
 
-    def add_info(self, labels: dict | None, data: dict) -> None:
+    def add_info(self, labels: Optional[dict[str, str]], data: dict) -> None:
         pass
 
-    def add_observation(self, labels: dict | None, value: float) -> None:
+    def add_observation(self, labels: Optional[dict[str, str]], value: float) -> None:
         pass
 
-    def increment_counter(self, labels: dict | None, amount: int = 1) -> None:
+    def increment_counter(self, labels: Optional[dict[str, str]], amount: int = 1) -> None:
         pass
 
-    def set_gauge(self, labels: dict, data: float) -> None:
+    def set_gauge(self, labels: Optional[dict[str, str]], data: float) -> None:
         pass
 
 
@@ -49,7 +50,7 @@ def __init__(
         self,
         name: str,
         description: str,
-        label_pairs: dict[str, str],
+        label_pairs: dict,
     ) -> None:
         """
         Args:
@@ -60,7 +61,9 @@ def __init__(
         super().__init__(name, description, label_pairs)
         self.counter = Counter(name, description, [*label_pairs.keys()])
 
-    def increment_counter(self, labels: dict[str, str] | None, amount: int = 1) -> None:
+    def increment_counter(self, labels: Optional[dict], amount: int = 1) -> None:
+        if not labels:
+            labels = {}
         _new_labels = self.label_pairs | labels
         self.counter.labels(**_new_labels).inc(amount=amount)
 
@@ -74,7 +77,7 @@ def __init__(
         self,
         name: str,
         description: str,
-        label_pairs: dict[str, str],
+        label_pairs: dict,
     ) -> None:
         """
         Args:
@@ -87,9 +90,11 @@ def __init__(
 
     def add_info(
         self,
-        labels: dict | None,
+        labels: Optional[dict[str, str]],
         data: dict,
     ) -> None:
+        if not labels:
+            labels = {}
         _new_labels = self.label_pairs | labels
         self.info.labels(**_new_labels).info(data)
 
@@ -101,7 +106,7 @@ def __init__(
         self,
         name: str,
         description: str,
-        label_pairs: dict[str, str],
+        label_pairs: dict,
     ) -> None:
         """
         Args:
@@ -112,7 +117,9 @@ def __init__(
         super().__init__(name, description, label_pairs)
         self.summary = Summary(name, description, [*label_pairs.keys()])
 
-    def add_observation(self, labels: dict | None, value: float) -> None:
+    def add_observation(self, labels: Optional[dict[str, str]], value: float) -> None:
+        if not labels:
+            labels = {}
         _new_labels = self.label_pairs | labels
         self.summary.labels(**_new_labels).observe(amount=value)
 
@@ -124,7 +131,7 @@ def __init__(
         self,
         name: str,
         description: str,
-        label_pairs: dict[str, str],
+        label_pairs: dict,
     ) -> None:
         """
         Args:
@@ -136,9 +143,11 @@ def __init__(
 
     def set_gauge(
         self,
-        labels: dict[str, str],
+        labels: Optional[dict[str, str]],
         data: float,
     ) -> None:
+        if not labels:
+            labels = {}
         _new_labels = self.label_pairs | labels
         self.gauge.labels(**_new_labels).set(data)
 
@@ -150,7 +159,7 @@ def __init__(
         self,
         name: str,
         description: str,
-        label_pairs: dict[str, str],
+        label_pairs: dict,
     ) -> None:
         """
         Args:
@@ -161,6 +170,8 @@ def __init__(
         super().__init__(name, description, label_pairs)
         self.histogram = Histogram(name, description, [*label_pairs.keys()])
 
-    def add_observation(self, labels: dict | None, value: float) -> None:
+    def add_observation(self, labels: Optional[dict[str, str]], value: float) -> None:
+        if not labels:
+            labels = {}
         _new_labels = self.label_pairs | labels
         self.histogram.labels(**_new_labels).observe(amount=value)

numaprom/monitoring/utility.py

@@ -11,7 +11,7 @@
 
 
 import logging
-
+from typing import Union, Optional
 
 from numaprom.monitoring.metrics import (
     PromCounterMetric,
@@ -29,16 +29,18 @@ def get_metric(
     metric_type: str,
     name: str,
     description: str,
-    label_pairs: dict[str, str] | None,
-) -> PromCounterMetric | PromInfoMetric | PromSummaryMetric | PromGaugeMetric | PromHistogramMetric:
+    label_pairs: Optional[dict[str, str]],
+) -> Union[
+    PromCounterMetric, PromInfoMetric, PromSummaryMetric, PromGaugeMetric, PromHistogramMetric
+]:
     """
     Returns a Prometheus metric object based on the metric type.
     Args:
         metric_type: metric type
         name: metric name
         description: metric description
         label_pairs: label pairs
-    Returns: _BaseMetric type
+    Returns: _BaseMetric covariant type
     """
     if metric_type == "Counter":
         return PromCounterMetric(name, description, label_pairs)