Adding timestamp to single_metric

Author: leandrodamascena

aws_lambda_powertools/metrics/base.py

@@ -557,6 +557,7 @@ def single_metric(
     resolution: MetricResolution | int = 60,
     namespace: str | None = None,
     default_dimensions: Dict[str, str] | None = None,
+    timestamp: int | datetime.datetime | None = None,
 ) -> Generator[SingleMetric, None, None]:
     """Context manager to simplify creation of a single metric
 
@@ -594,6 +595,9 @@ def single_metric(
         Metric value
     namespace: str
         Namespace for metrics
+    default_dimensions: Dict[str, str], optional
+        Metric dimensions as key=value that will always be present
+
 
     Yields
     -------
@@ -616,6 +620,9 @@ def single_metric(
         metric: SingleMetric = SingleMetric(namespace=namespace)
         metric.add_metric(name=name, unit=unit, value=value, resolution=resolution)
 
+        if timestamp:
+            metric.set_timestamp(timestamp)
+
         if default_dimensions:
             for dim_name, dim_value in default_dimensions.items():
                 metric.add_dimension(name=dim_name, value=dim_value)

tests/functional/metrics/test_metrics_cloudwatch_emf.py

@@ -57,6 +57,7 @@ def serialize_single_metric(
     dimension: Dict,
     namespace: str,
     metadata: Dict | None = None,
+    timestamp: int | datetime.datetime | None = None,
 ) -> CloudWatchEMFOutput:
     """Helper function to build EMF object from a given metric, dimension and namespace"""
     my_metrics = AmazonCloudWatchEMFProvider(namespace=namespace)
@@ -66,6 +67,9 @@ def serialize_single_metric(
     if metadata is not None:
         my_metrics.add_metadata(**metadata)
 
+    if timestamp:
+        my_metrics.set_timestamp(timestamp)
+
     return my_metrics.serialize_metric_set()
 
 
@@ -143,6 +147,28 @@ def test_single_metric_default_dimensions(capsys, metric, dimension, namespace):
     assert expected == output
 
 
+def test_single_metric_with_custom_timestamp(capsys, metric, dimension, namespace):
+    # GIVEN we provide a custom timestamp
+    # WHEN using single_metric context manager
+
+    default_dimensions = {dimension["name"]: dimension["value"]}
+
+    timestamp = int((datetime.datetime.now() - datetime.timedelta(days=2)).timestamp() * 1000)
+    with single_metric(
+        namespace=namespace,
+        default_dimensions=default_dimensions,
+        **metric,
+        timestamp=timestamp,
+    ) as my_metric:
+        my_metric.add_metric(name="second_metric", unit="Count", value=1)
+
+    output = capture_metrics_output(capsys)
+    expected = serialize_single_metric(metric=metric, dimension=dimension, namespace=namespace, timestamp=timestamp)
+
+    # THEN we should have custom timestamp added to the metric
+    assert expected == output
+
+
 def test_single_metric_default_dimensions_inherit(capsys, metric, dimension, namespace):
     # GIVEN we provide Metrics default dimensions
     # WHEN using single_metric context manager