chore: Add type to metrics code and deprecate CWMetricsPublisher

samtranslator/internal/deprecation_control.py

@@ -23,7 +23,10 @@ def _make_message(message: str, replacement: Optional[str]) -> str:
     return f"{message}, please use {replacement}" if replacement else message
 
 
-def deprecated(replacement: Optional[str]) -> Callable[[Callable[PT, RT]], Callable[PT, RT]]:
+# TODO: make @deprecated able to decorate a class
+
+
+def deprecated(replacement: Optional[str] = None) -> Callable[[Callable[PT, RT]], Callable[PT, RT]]:
     """
     Mark a function/method as deprecated.
 

samtranslator/metrics/method_decorator.py

@@ -76,7 +76,7 @@ def _send_cw_metric(prefix, name, execution_time_ms, func, args):  # type: ignor
     try:
         metric_name = _get_metric_name(prefix, name, func, args)  # type: ignore[no-untyped-call]
         LOG.debug("Execution took %sms for %s", execution_time_ms, metric_name)
-        MetricsMethodWrapperSingleton.get_instance().record_latency(metric_name, execution_time_ms)  # type: ignore[no-untyped-call]
+        MetricsMethodWrapperSingleton.get_instance().record_latency(metric_name, execution_time_ms)
     except Exception as e:
         LOG.warning("Failed to add metrics", exc_info=e)
 

samtranslator/metrics/metrics.py

@@ -4,7 +4,11 @@
 import logging
 from abc import ABC, abstractmethod
 from datetime import datetime
-from typing import Any, Dict, List, Optional
+from typing import Any, Dict, List, Optional, Union
+
+from typing_extensions import TypedDict
+
+from samtranslator.internal.deprecation_control import deprecated
 
 LOG = logging.getLogger(__name__)
 
@@ -25,6 +29,7 @@ def publish(self, namespace: str, metrics: List["MetricDatum"]) -> None:
 class CWMetricsPublisher(MetricsPublisher):
     BATCH_SIZE = 20
 
+    @deprecated()
     def __init__(self, cloudwatch_client) -> None:  # type: ignore[no-untyped-def]
         """
         Constructor
@@ -66,7 +71,7 @@ class DummyMetricsPublisher(MetricsPublisher):
     def __init__(self) -> None:
         MetricsPublisher.__init__(self)
 
-    def publish(self, namespace, metrics):  # type: ignore[no-untyped-def]
+    def publish(self, namespace: str, metrics: List["MetricDatum"]) -> None:
         """Do not publish any metric, this is a dummy publisher used for offline use."""
         LOG.debug(f"Dummy publisher ignoring {len(metrics)} metrices")
 
@@ -90,7 +95,14 @@ class MetricDatum:
     Class to hold Metric data.
     """
 
-    def __init__(self, name, value, unit, dimensions=None, timestamp=None) -> None:  # type: ignore[no-untyped-def]
+    def __init__(
+        self,
+        name: str,
+        value: Union[int, float],
+        unit: str,
+        dimensions: Optional[List["MetricDimension"]] = None,
+        timestamp: Optional[datetime] = None,
+    ) -> None:
         """
         Constructor
 
@@ -116,6 +128,11 @@ def get_metric_data(self) -> Dict[str, Any]:
         }
 
 
+class MetricDimension(TypedDict):
+    Name: str
+    Value: Any
+
+
 class Metrics:
     def __init__(
         self, namespace: str = "ServerlessTransform", metrics_publisher: Optional[MetricsPublisher] = None
@@ -138,7 +155,14 @@ def __del__(self) -> None:
             )
             self.publish()
 
-    def _record_metric(self, name, value, unit, dimensions=None, timestamp=None):  # type: ignore[no-untyped-def]
+    def _record_metric(
+        self,
+        name: str,
+        value: Union[int, float],
+        unit: str,
+        dimensions: Optional[List["MetricDimension"]] = None,
+        timestamp: Optional[datetime] = None,
+    ) -> None:
         """
         Create and save metric object in internal cache.
 
@@ -150,7 +174,13 @@ def _record_metric(self, name, value, unit, dimensions=None, timestamp=None):  #
         """
         self.metrics_cache.setdefault(name, []).append(MetricDatum(name, value, unit, dimensions, timestamp))
 
-    def record_count(self, name, value, dimensions=None, timestamp=None):  # type: ignore[no-untyped-def]
+    def record_count(
+        self,
+        name: str,
+        value: int,
+        dimensions: Optional[List["MetricDimension"]] = None,
+        timestamp: Optional[datetime] = None,
+    ) -> None:
         """
         Create metric with unit Count.
 
@@ -160,9 +190,15 @@ def record_count(self, name, value, dimensions=None, timestamp=None):  # type: i
         :param dimensions: array of dimensions applied to the metric
         :param timestamp: timestamp of metric (datetime.datetime object)
         """
-        self._record_metric(name, value, Unit.Count, dimensions, timestamp)  # type: ignore[no-untyped-call]
+        self._record_metric(name, value, Unit.Count, dimensions, timestamp)
 
-    def record_latency(self, name, value, dimensions=None, timestamp=None):  # type: ignore[no-untyped-def]
+    def record_latency(
+        self,
+        name: str,
+        value: Union[int, float],
+        dimensions: Optional[List["MetricDimension"]] = None,
+        timestamp: Optional[datetime] = None,
+    ) -> None:
         """
         Create metric with unit Milliseconds.
 
@@ -172,7 +208,7 @@ def record_latency(self, name, value, dimensions=None, timestamp=None):  # type:
         :param dimensions: array of dimensions applied to the metric
         :param timestamp: timestamp of metric (datetime.datetime object)
         """
-        self._record_metric(name, value, Unit.Milliseconds, dimensions, timestamp)  # type: ignore[no-untyped-call]
+        self._record_metric(name, value, Unit.Milliseconds, dimensions, timestamp)
 
     def publish(self) -> None:
         """Calls publish method from the configured metrics publisher to publish metrics"""
@@ -184,7 +220,7 @@ def publish(self) -> None:
         self.metrics_publisher.publish(self.namespace, all_metrics)
         self.metrics_cache = {}
 
-    def get_metric(self, name):  # type: ignore[no-untyped-def]
+    def get_metric(self, name: str) -> List[MetricDatum]:
         """
         Returns a list of metrics from the internal cache for a metric name
 

samtranslator/model/eventsources/pull.py

@@ -659,7 +659,7 @@ def get_vpc_permission(self) -> Dict[str, Any]:
         }
 
     @staticmethod
-    @deprecated(None)
+    @deprecated()
     def get_kms_policy(secrets_manager_kms_key_id: str) -> Dict[str, Any]:
         return {
             "Action": ["kms:Decrypt"],