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

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/metrics.py

@@ -138,9 +138,6 @@ def log_metrics(
             default_dimensions=default_dimensions,
         )
 
-    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

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",
 ]

aws_lambda_powertools/metrics/provider/base.py

@@ -2,19 +2,17 @@
 
 import functools
 import logging
+from abc import ABC, abstractmethod
 from typing import Any, Callable, Dict, Optional
 
-from typing_extensions import Protocol, runtime_checkable
+from aws_lambda_powertools.metrics.provider import cold_start
 
 logger = logging.getLogger(__name__)
 
-is_cold_start = True
 
-
-@runtime_checkable
-class MetricsProviderBase(Protocol):
+class BaseProvider(ABC):
     """
-    Interface for MetricsProvider.
+    Class for metric provider interface.
 
     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,41 +23,127 @@ class MetricsProviderBase(Protocol):
         3. Customize the behavior and functionality of the metric provider in your subclass.
     """
 
+    @abstractmethod
     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
+
+    @abstractmethod
     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
+
+    @abstractmethod
     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.
 
-@runtime_checkable
-class MetricsBase(Protocol):
-    """
-    Interface for metric template.
+        Parameters
+        ----------
+        *args:
+            Positional arguments.
+        *kwargs:
+            Keyword arguments.
 
-    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.
+        Raises
+        ----------
+        NotImplementedError
+            This method must be implemented in subclasses.
+        """
+        raise NotImplementedError
 
-    NOTE: need to improve this docstring
-    """
+    @abstractmethod
+    def clear_metrics(self, *args: Any, **kwargs) -> Any:
+        """
+        Abstract method for clear metric instance.
+
+        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
 
-    def add_metric(self, *args, **kwargs):
-        ...
+    @abstractmethod
+    def add_cold_start_metric(self, context: Any) -> Any:
+        """
+        Abstract method for clear metric instance.
+
+        This method must be implemented in subclasses to add a metric and return a combined metrics dictionary.
 
-    def flush_metrics(self, raise_on_empty_metrics: bool = False) -> None:
-        ...
+        Parameters
+        ----------
+        *args:
+            Positional arguments.
+        *kwargs:
+            Keyword arguments.
 
-    def add_cold_start_metric(self, metric_name: str, function_name: str) -> None:
-        ...
+        Raises
+        ----------
+        NotImplementedError
+            This method must be implemented in subclasses.
+        """
+        raise NotImplementedError
 
     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,
+        **kwargs,
     ):
         """Decorator to serialize and publish metrics at the end of a function execution.
 
@@ -96,6 +180,8 @@ def handler(event, context):
             Propagate error received
         """
 
+        default_dimensions = kwargs.get("default_dimensions")
+
         # If handler is None we've been called with parameters
         # Return a partial function with args filled
         if lambda_handler is None:
@@ -104,11 +190,14 @@ def handler(event, context):
                 self.log_metrics,
                 capture_cold_start_metric=capture_cold_start_metric,
                 raise_on_empty_metrics=raise_on_empty_metrics,
+                default_dimensions=default_dimensions,
             )
 
         @functools.wraps(lambda_handler)
         def decorate(event, context):
             try:
+                if default_dimensions:
+                    self.set_default_dimensions(**default_dimensions)
                 response = lambda_handler(event, context)
                 if capture_cold_start_metric:
                     self._add_cold_start_metric(context=context)
@@ -120,24 +209,23 @@ def decorate(event, context):
         return decorate
 
     def _add_cold_start_metric(self, context: Any) -> None:
-        """Add cold start metric and function_name dimension
+        """
+        Check if it's cold start and add a metric if yes
 
         Parameters
         ----------
         context : Any
             Lambda context
         """
-        global is_cold_start
-        if not is_cold_start:
+        if not cold_start.is_cold_start:
             return
 
         logger.debug("Adding cold start metric and function_name dimension")
-        self.add_cold_start_metric(metric_name="ColdStart", function_name=context.function_name)
+        self.add_cold_start_metric(context=context)
 
-        is_cold_start = False
+        cold_start.is_cold_start = False
 
 
 def reset_cold_start_flag_provider():
-    global is_cold_start
-    if not is_cold_start:
-        is_cold_start = True
+    if not cold_start.is_cold_start:
+        cold_start.is_cold_start = True

aws_lambda_powertools/metrics/provider/cloudwatch_emf/cloudwatch.py

@@ -1,7 +1,6 @@
 from __future__ import annotations
 
 import datetime
-import functools
 import json
 import logging
 import numbers
@@ -12,30 +11,31 @@
 
 from aws_lambda_powertools.metrics.base import single_metric
 from aws_lambda_powertools.metrics.exceptions import MetricValueError, SchemaValidationError
-from aws_lambda_powertools.metrics.provider.cloudwatch_emf 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.shared import (
+from aws_lambda_powertools.metrics.functions import (
     extract_cloudwatch_metric_resolution_value,
     extract_cloudwatch_metric_unit_value,
 )
+from aws_lambda_powertools.metrics.provider.base import BaseProvider
+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.types import MetricNameUnitResolution
 from aws_lambda_powertools.shared import constants
 from aws_lambda_powertools.shared.functions import resolve_env_var_choice
 
 logger = logging.getLogger(__name__)
 
 
-class AmazonCloudWatchEMFProvider:
-    """Base class for metric functionality (namespace, metric, dimension, serialization)
+class AmazonCloudWatchEMFProvider(BaseProvider):
+    """
+    AmazonCloudWatchEMFProvider class (namespace, metric, dimension, serialization)
 
-    MetricManager creates metrics asynchronously thanks to CloudWatch Embedded Metric Format (EMF).
+    AmazonCloudWatchEMFProvider creates metrics asynchronously thanks to CloudWatch Embedded Metric Format (EMF).
     CloudWatch EMF can create up to 100 metrics per EMF object
-    and metrics, dimensions, and namespace created via MetricManager
+    and metrics, dimensions, and namespace created via AmazonCloudWatchEMFProvider
     will adhere to the schema, will be serialized and validated against EMF Schema.
 
-    **Use `aws_lambda_powertools.metrics.metrics.Metrics` or
-    `aws_lambda_powertools.metrics.metric.single_metric` to create EMF metrics.**
+    **Use `aws_lambda_powertools.Metrics` or
+    `aws_lambda_powertools.single_metric` to create EMF metrics.**
 
     Environment variables
     ---------------------
@@ -335,7 +335,7 @@ def log_metrics(
         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,
+        **kwargs,
     ):
         """Decorator to serialize and publish metrics at the end of a function execution.
 
@@ -363,56 +363,34 @@ def handler(event, context):
             captures cold start metric, by default False
         raise_on_empty_metrics : bool, optional
             raise exception if no metrics are emitted, by default False
-        default_dimensions: Dict[str, str], optional
-            metric dimensions as key=value that will always be present
+        **kwargs
 
         Raises
         ------
         e
             Propagate error received
         """
 
-        # If handler is None we've been called with parameters
-        # Return a partial function with args filled
-        if lambda_handler is None:
-            logger.debug("Decorator called with parameters")
-            return functools.partial(
-                self.log_metrics,
-                capture_cold_start_metric=capture_cold_start_metric,
-                raise_on_empty_metrics=raise_on_empty_metrics,
-                default_dimensions=default_dimensions,
-            )
-
-        @functools.wraps(lambda_handler)
-        def decorate(event, context):
-            try:
-                if default_dimensions:
-                    self.set_default_dimensions(**default_dimensions)
-                response = lambda_handler(event, context)
-                if capture_cold_start_metric:
-                    self._add_cold_start_metric(context=context)
-            finally:
-                self.flush_metrics(raise_on_empty_metrics=raise_on_empty_metrics)
-
-            return response
-
-        return decorate
+        return super().log_metrics(
+            lambda_handler=lambda_handler,
+            capture_cold_start_metric=capture_cold_start_metric,
+            raise_on_empty_metrics=raise_on_empty_metrics,
+            **kwargs,
+        )
 
-    def _add_cold_start_metric(self, context: Any) -> None:
+    def add_cold_start_metric(self, context: Any) -> None:
         """Add cold start metric and function_name dimension
 
         Parameters
         ----------
         context : Any
             Lambda context
         """
-        if cold_start.is_cold_start:
-            logger.debug("Adding cold start metric and function_name dimension")
-            with single_metric(name="ColdStart", unit=MetricUnit.Count, value=1, namespace=self.namespace) as metric:
-                metric.add_dimension(name="function_name", value=context.function_name)
-                if self.service:
-                    metric.add_dimension(name="service", value=str(self.service))
-                cold_start.is_cold_start = False
+        logger.debug("Adding cold start metric and function_name dimension")
+        with single_metric(name="ColdStart", unit=MetricUnit.Count, value=1, namespace=self.namespace) as metric:
+            metric.add_dimension(name="function_name", value=context.function_name)
+            if self.service:
+                metric.add_dimension(name="service", value=str(self.service))
 
     def set_default_dimensions(self, **dimensions) -> None:
         """Persist dimensions across Lambda invocations