refactor(metrics): move from protocol to ABC; split provider tests (#2934)

Author: leandrodamascena

aws_lambda_powertools/metrics/base.py

@@ -17,12 +17,12 @@
     MetricValueError,
     SchemaValidationError,
 )
-from aws_lambda_powertools.metrics.provider.cloudwatch_emf import cold_start
-from aws_lambda_powertools.metrics.provider.cloudwatch_emf.cold_start import (
-    reset_cold_start_flag,  # noqa: F401  # backwards compatibility
-)
+from aws_lambda_powertools.metrics.provider import cold_start
 from aws_lambda_powertools.metrics.provider.cloudwatch_emf.constants import MAX_DIMENSIONS, MAX_METRICS
 from aws_lambda_powertools.metrics.provider.cloudwatch_emf.metric_properties import MetricResolution, MetricUnit
+from aws_lambda_powertools.metrics.provider.cold_start import (
+    reset_cold_start_flag,  # noqa: F401  # backwards compatibility
+)
 from aws_lambda_powertools.metrics.types import MetricNameUnitResolution
 from aws_lambda_powertools.shared import constants
 from aws_lambda_powertools.shared.functions import resolve_env_var_choice

aws_lambda_powertools/metrics/functions.py

@@ -0,0 +1,72 @@
+from __future__ import annotations
+
+from typing import List
+
+from aws_lambda_powertools.metrics.provider.cloudwatch_emf.exceptions import (
+    MetricResolutionError,
+    MetricUnitError,
+)
+from aws_lambda_powertools.metrics.provider.cloudwatch_emf.metric_properties import MetricResolution, MetricUnit
+
+
+def extract_cloudwatch_metric_resolution_value(metric_resolutions: List, resolution: int | MetricResolution) -> int:
+    """Return metric value from CloudWatch metric unit whether that's str or MetricResolution enum
+
+    Parameters
+    ----------
+    unit : Union[int, MetricResolution]
+        Metric resolution
+
+    Returns
+    -------
+    int
+        Metric resolution value must be 1 or 60
+
+    Raises
+    ------
+    MetricResolutionError
+        When metric resolution is not supported by CloudWatch
+    """
+    if isinstance(resolution, MetricResolution):
+        return resolution.value
+
+    if isinstance(resolution, int) and resolution in metric_resolutions:
+        return resolution
+
+    raise MetricResolutionError(
+        f"Invalid metric resolution '{resolution}', expected either option: {metric_resolutions}",  # noqa: E501
+    )
+
+
+def extract_cloudwatch_metric_unit_value(metric_units: List, metric_valid_options: List, unit: str | MetricUnit) -> str:
+    """Return metric value from CloudWatch metric unit whether that's str or MetricUnit enum
+
+    Parameters
+    ----------
+    unit : Union[str, MetricUnit]
+        Metric unit
+
+    Returns
+    -------
+    str
+        Metric unit value (e.g. "Seconds", "Count/Second")
+
+    Raises
+    ------
+    MetricUnitError
+        When metric unit is not supported by CloudWatch
+    """
+
+    if isinstance(unit, str):
+        if unit in metric_valid_options:
+            unit = MetricUnit[unit].value
+
+        if unit not in metric_units:
+            raise MetricUnitError(
+                f"Invalid metric unit '{unit}', expected either option: {metric_valid_options}",
+            )
+
+    if isinstance(unit, MetricUnit):
+        unit = unit.value
+
+    return unit

aws_lambda_powertools/metrics/metrics.py

@@ -1,14 +1,15 @@
 # NOTE: keeps for compatibility
 from __future__ import annotations
 
-from typing import Any, Callable, Dict, Optional, Union
+from typing import Any, Callable, Dict, Optional
 
 from aws_lambda_powertools.metrics.base import MetricResolution, MetricUnit
 from aws_lambda_powertools.metrics.provider.cloudwatch_emf.cloudwatch import AmazonCloudWatchEMFProvider
+from aws_lambda_powertools.metrics.provider.cloudwatch_emf.types import CloudWatchEMFOutput
 
 
 class Metrics:
-    """Metrics create an EMF object with up to 100 metrics
+    """Metrics create an CloudWatch EMF object with up to 100 metrics
 
     Use Metrics when you need to create multiple metrics that have
     dimensions in common (e.g. service_name="payment").
@@ -115,7 +116,7 @@ def serialize_metric_set(
         metrics: Dict | None = None,
         dimensions: Dict | None = None,
         metadata: Dict | None = None,
-    ) -> Dict:
+    ) -> CloudWatchEMFOutput:
         return self.provider.serialize_metric_set(metrics=metrics, dimensions=dimensions, metadata=metadata)
 
     def add_metadata(self, key: str, value: Any) -> None:
@@ -138,15 +139,6 @@ def log_metrics(
             default_dimensions=default_dimensions,
         )
 
-    def _extract_metric_resolution_value(self, resolution: Union[int, MetricResolution]) -> int:
-        return self.provider._extract_metric_resolution_value(resolution=resolution)
-
-    def _extract_metric_unit_value(self, unit: Union[str, MetricUnit]) -> str:
-        return self.provider._extract_metric_unit_value(unit=unit)
-
-    def _add_cold_start_metric(self, context: Any) -> None:
-        self.provider._add_cold_start_metric(context=context)
-
     def set_default_dimensions(self, **dimensions) -> None:
         self.provider.set_default_dimensions(**dimensions)
         """Persist dimensions across Lambda invocations
@@ -208,120 +200,3 @@ def service(self, service):
 # we can quickly revert and duplicate code while using self.provider
 
 EphemeralMetrics = AmazonCloudWatchEMFProvider
-
-# noqa: ERA001
-# class EphemeralMetrics(MetricManager):
-#     """Non-singleton version of Metrics to not persist metrics across instances
-#
-#     NOTE: This is useful when you want to:
-#
-#     - Create metrics for distinct namespaces
-#     - Create the same metrics with different dimensions more than once
-#     """
-#
-#     # _dimensions: Dict[str, str] = {}
-#     _default_dimensions: Dict[str, Any] = {}
-#
-#     def __init__(
-#         self,
-#         service: str | None = None,
-#         namespace: str | None = None,
-#         provider: AmazonCloudWatchEMFProvider | None = None,
-#     ):
-#         super().__init__(namespace=namespace, service=service)
-#
-#         self.default_dimensions = self._default_dimensions
-#         # # self.dimension_set = self._dimensions
-#         # self.dimension_set.update(**self._default_dimensions)
-#
-#         self.provider = provider or AmazonCloudWatchEMFProvider(
-#             namespace=namespace,
-#             service=service,
-#             metric_set=self.metric_set,
-#             metadata_set=self.metadata_set,
-#             dimension_set=self.dimension_set,
-#             default_dimensions=self._default_dimensions,
-#         )
-#
-#     def add_metric(
-#         self,
-#         name: str,
-#         unit: MetricUnit | str,
-#         value: float,
-#         resolution: MetricResolution | int = 60,
-#     ) -> None:
-#         return self.provider.add_metric(name=name, unit=unit, value=value, resolution=resolution)
-#
-#     def add_dimension(self, name: str, value: str) -> None:
-#         return self.provider.add_dimension(name=name, value=value)
-#
-#     def serialize_metric_set(
-#         self,
-#         metrics: Dict | None = None,
-#         dimensions: Dict | None = None,
-#         metadata: Dict | None = None,
-#     ) -> Dict:
-#         return self.provider.serialize_metric_set(metrics=metrics, dimensions=dimensions, metadata=metadata)
-#
-#     def add_metadata(self, key: str, value: Any) -> None:
-#         self.provider.add_metadata(key=key, value=value)
-#
-#     def flush_metrics(self, raise_on_empty_metrics: bool = False) -> None:
-#         self.provider.flush_metrics(raise_on_empty_metrics=raise_on_empty_metrics)
-#
-#     def log_metrics(
-#         self,
-#         lambda_handler: Callable[[Dict, Any], Any] | Optional[Callable[[Dict, Any, Optional[Dict]], Any]] = None,
-#         capture_cold_start_metric: bool = False,
-#         raise_on_empty_metrics: bool = False,
-#         default_dimensions: Dict[str, str] | None = None,
-#     ):
-#         return self.provider.log_metrics(
-#             lambda_handler=lambda_handler,
-#             capture_cold_start_metric=capture_cold_start_metric,
-#             raise_on_empty_metrics=raise_on_empty_metrics,
-#             default_dimensions=default_dimensions,
-#         )
-#
-#     def _extract_metric_resolution_value(self, resolution: Union[int, MetricResolution]) -> int:
-#         return self.provider._extract_metric_resolution_value(resolution=resolution)
-#
-#     def _extract_metric_unit_value(self, unit: Union[str, MetricUnit]) -> str:
-#         return self.provider._extract_metric_unit_value(unit=unit)
-#
-#     def _add_cold_start_metric(self, context: Any) -> None:
-#         return self.provider._add_cold_start_metric(context=context)
-#
-#     def set_default_dimensions(self, **dimensions) -> None:
-#         """Persist dimensions across Lambda invocations
-#
-#         Parameters
-#         ----------
-#         dimensions : Dict[str, Any], optional
-#             metric dimensions as key=value
-#
-#         Example
-#         -------
-#         **Sets some default dimensions that will always be present across metrics and invocations**
-#
-#             from aws_lambda_powertools import Metrics
-#
-#             metrics = Metrics(namespace="ServerlessAirline", service="payment")
-#             metrics.set_default_dimensions(environment="demo", another="one")
-#
-#             @metrics.log_metrics()
-#             def lambda_handler():
-#                 return True
-#         """
-#         return self.provider.set_default_dimensions(**dimensions)
-#
-#     def clear_default_dimensions(self) -> None:
-#         self.default_dimensions.clear()
-#
-#     def clear_metrics(self) -> None:
-#         self.provider.clear_metrics()
-#         # re-add default dimensions
-#         self.set_default_dimensions(**self.default_dimensions)
-#
-
-# __all__ = []

aws_lambda_powertools/metrics/provider/__init__.py

@@ -1,6 +1,5 @@
-from aws_lambda_powertools.metrics.provider.base import MetricsBase, MetricsProviderBase
+from aws_lambda_powertools.metrics.provider.base import BaseProvider
 
 __all__ = [
-    "MetricsBase",
-    "MetricsProviderBase",
+    "BaseProvider",
 ]