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

aws_lambda_powertools/metrics/metrics.py

@@ -1,14 +1,14 @@
 # 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
 
 
 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").
@@ -138,12 +138,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)
 
@@ -208,120 +202,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/base.py

@@ -4,16 +4,17 @@
 import logging
 from typing import Any, Callable, Dict, Optional
 
-from typing_extensions import Protocol
+from typing_extensions import Protocol, runtime_checkable
 
 logger = logging.getLogger(__name__)
 
 is_cold_start = True
 
 
+@runtime_checkable
 class MetricsProviderBase(Protocol):
     """
-    Class for metric provider interface.
+    Interface for MetricsProvider.
 
     This class serves as an interface for creating your own metric provider. Inherit from this class
     and implement the required methods to define your specific metric provider.
@@ -25,80 +26,19 @@ class MetricsProviderBase(Protocol):
     """
 
     def add_metric(self, *args: Any, **kwargs: Any) -> Any:
-        """
-        Abstract method for adding a metric.
-
-        This method must be implemented in subclasses to add a metric and return a combined metrics dictionary.
-
-        Parameters
-        ----------
-        *args:
-            Positional arguments.
-        *kwargs:
-            Keyword arguments.
-
-        Returns
-        ----------
-        Dict
-            A combined metrics dictionary.
-
-        Raises
-        ----------
-        NotImplementedError
-            This method must be implemented in subclasses.
-        """
-        raise NotImplementedError
+        ...
 
     def serialize_metric_set(self, *args: Any, **kwargs: Any) -> Any:
-        """
-        Abstract method for serialize a metric.
-
-        This method must be implemented in subclasses to add a metric and return a combined metrics dictionary.
+        ...
 
-        Parameters
-        ----------
-        *args:
-            Positional arguments.
-        *kwargs:
-            Keyword arguments.
-
-        Returns
-        ----------
-        Dict
-            Serialized metrics
-
-        Raises
-        ----------
-        NotImplementedError
-            This method must be implemented in subclasses.
-        """
-        raise NotImplementedError
-
-    # flush serialized data to output, or send to API directly
     def flush_metrics(self, *args: Any, **kwargs) -> Any:
-        """
-        Abstract method for flushing a metric.
-
-        This method must be implemented in subclasses to add a metric and return a combined metrics dictionary.
-
-        Parameters
-        ----------
-        *args:
-            Positional arguments.
-        *kwargs:
-            Keyword arguments.
-
-        Raises
-        ----------
-        NotImplementedError
-            This method must be implemented in subclasses.
-        """
-        raise NotImplementedError
+        ...
 
 
+@runtime_checkable
 class MetricsBase(Protocol):
     """
-    Class for metric template.
+    Interface for metric template.
 
     This class serves as a template for creating your own metric class. Inherit from this class
     and implement the necessary methods to define your specific metric.
@@ -107,54 +47,13 @@ class MetricsBase(Protocol):
     """
 
     def add_metric(self, *args, **kwargs):
-        """
-        Abstract method for adding a metric.
-
-        This method must be implemented in subclasses to add a metric and return a combined metrics dictionary.
-
-        Parameters
-        ----------
-        *args:
-            Positional arguments.
-        *kwargs:
-            Keyword arguments.
-
-        Returns
-        ----------
-        Dict
-            A combined metrics dictionary.
-
-        Raises
-        ----------
-        NotImplementedError
-            This method must be implemented in subclasses.
-        """
-        raise NotImplementedError
+        ...
 
     def flush_metrics(self, raise_on_empty_metrics: bool = False) -> None:
-        """Manually flushes the metrics. This is normally not necessary,
-        unless you're running on other runtimes besides Lambda, where the @log_metrics
-        decorator already handles things for you.
-
-        Parameters
-        ----------
-        raise_on_empty_metrics : bool, optional
-            raise exception if no metrics are emitted, by default False
-        """
-        raise NotImplementedError
+        ...
 
     def add_cold_start_metric(self, metric_name: str, function_name: str) -> None:
-        """
-        Add a cold start metric for a specific function.
-
-        Parameters
-        ----------
-        metric_name: str
-            The name of the cold start metric to add.
-        function_name: str
-            The name of the function associated with the cold start metric.
-        """
-        raise NotImplementedError
+        ...
 
     def log_metrics(
         self,