refactor(metrics): add from __future__ import annotations (#4944)

and update code according to ruff rules TCH, UP006, UP007, UP037 and
FA100.

Author: ericbn

aws_lambda_powertools/metrics/base.py

@@ -9,7 +9,7 @@
 import warnings
 from collections import defaultdict
 from contextlib import contextmanager
-from typing import Any, Callable, Dict, Generator, List, Optional, Union
+from typing import TYPE_CHECKING, Any, Callable, Generator
 
 from aws_lambda_powertools.metrics.exceptions import (
     MetricResolutionError,
@@ -24,10 +24,12 @@
 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
 
+if TYPE_CHECKING:
+    from aws_lambda_powertools.metrics.types import MetricNameUnitResolution
+
 logger = logging.getLogger(__name__)
 
 # Maintenance: alias due to Hyrum's law
@@ -66,10 +68,10 @@ class MetricManager:
 
     def __init__(
         self,
-        metric_set: Dict[str, Any] | None = None,
-        dimension_set: Dict | None = None,
+        metric_set: dict[str, Any] | None = None,
+        dimension_set: dict | None = None,
         namespace: str | None = None,
-        metadata_set: Dict[str, Any] | None = None,
+        metadata_set: dict[str, Any] | None = None,
         service: str | None = None,
     ):
         self.metric_set = metric_set if metric_set is not None else {}
@@ -110,11 +112,11 @@ def add_metric(
         ----------
         name : str
             Metric name
-        unit : Union[MetricUnit, str]
+        unit : MetricUnit | str
             `aws_lambda_powertools.helper.models.MetricUnit`
         value : float
             Metric value
-        resolution : Union[MetricResolution, int]
+        resolution : MetricResolution | int
             `aws_lambda_powertools.helper.models.MetricResolution`
 
         Raises
@@ -129,7 +131,7 @@ def add_metric(
 
         unit = self._extract_metric_unit_value(unit=unit)
         resolution = self._extract_metric_resolution_value(resolution=resolution)
-        metric: Dict = self.metric_set.get(name, defaultdict(list))
+        metric: dict = self.metric_set.get(name, defaultdict(list))
         metric["Unit"] = unit
         metric["StorageResolution"] = resolution
         metric["Value"].append(float(value))
@@ -147,19 +149,19 @@ def add_metric(
 
     def serialize_metric_set(
         self,
-        metrics: Dict | None = None,
-        dimensions: Dict | None = None,
-        metadata: Dict | None = None,
-    ) -> Dict:
+        metrics: dict | None = None,
+        dimensions: dict | None = None,
+        metadata: dict | None = None,
+    ) -> dict:
         """Serializes metric and dimensions set
 
         Parameters
         ----------
-        metrics : Dict, optional
+        metrics : dict, optional
             Dictionary of metrics to serialize, by default None
-        dimensions : Dict, optional
+        dimensions : dict, optional
             Dictionary of dimensions to serialize, by default None
-        metadata: Dict, optional
+        metadata: dict, optional
             Dictionary of metadata to serialize, by default None
 
         Example
@@ -172,7 +174,7 @@ def serialize_metric_set(
 
         Returns
         -------
-        Dict
+        dict
             Serialized metrics following EMF specification
 
         Raises
@@ -206,8 +208,8 @@ def serialize_metric_set(
         #
         # In case using high-resolution metrics, add StorageResolution field
         # Example: [ { "Name": "metric_name", "Unit": "Count", "StorageResolution": 1 } ] # noqa ERA001
-        metric_definition: List[MetricNameUnitResolution] = []
-        metric_names_and_values: Dict[str, float] = {}  # { "metric_name": 1.0 }
+        metric_definition: list[MetricNameUnitResolution] = []
+        metric_names_and_values: dict[str, float] = {}  # { "metric_name": 1.0 }
 
         for metric_name in metrics:
             metric: dict = metrics[metric_name]
@@ -354,10 +356,10 @@ def flush_metrics(self, raise_on_empty_metrics: bool = False) -> None:
 
     def log_metrics(
         self,
-        lambda_handler: Callable[[Dict, Any], Any] | Optional[Callable[[Dict, Any, Optional[Dict]], Any]] = None,
+        lambda_handler: Callable[[dict, Any], Any] | Callable[[dict, Any, dict | None], Any] | None = None,
         capture_cold_start_metric: bool = False,
         raise_on_empty_metrics: bool = False,
-        default_dimensions: Dict[str, str] | None = None,
+        default_dimensions: dict[str, str] | None = None,
     ):
         """Decorator to serialize and publish metrics at the end of a function execution.
 
@@ -385,7 +387,7 @@ 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
+        default_dimensions: dict[str, str], optional
             metric dimensions as key=value that will always be present
 
         Raises
@@ -420,12 +422,12 @@ def decorate(event, context, *args, **kwargs):
 
         return decorate
 
-    def _extract_metric_resolution_value(self, resolution: Union[int, MetricResolution]) -> int:
+    def _extract_metric_resolution_value(self, resolution: int | MetricResolution) -> int:
         """Return metric value from metric unit whether that's str or MetricResolution enum
 
         Parameters
         ----------
-        unit : Union[int, MetricResolution]
+        unit : int | MetricResolution
             Metric resolution
 
         Returns
@@ -448,12 +450,12 @@ def _extract_metric_resolution_value(self, resolution: Union[int, MetricResoluti
             f"Invalid metric resolution '{resolution}', expected either option: {self._metric_resolutions}",  # noqa: E501
         )
 
-    def _extract_metric_unit_value(self, unit: Union[str, MetricUnit]) -> str:
+    def _extract_metric_unit_value(self, unit: str | MetricUnit) -> str:
         """Return metric value from metric unit whether that's str or MetricUnit enum
 
         Parameters
         ----------
-        unit : Union[str, MetricUnit]
+        unit : str | MetricUnit
             Metric unit
 
         Returns
@@ -566,7 +568,7 @@ def single_metric(
     value: float,
     resolution: MetricResolution | int = 60,
     namespace: str | None = None,
-    default_dimensions: Dict[str, str] | None = None,
+    default_dimensions: dict[str, str] | None = None,
 ) -> Generator[SingleMetric, None, None]:
     """Context manager to simplify creation of a single metric
 
@@ -604,7 +606,7 @@ def single_metric(
         Metric value
     namespace: str
         Namespace for metrics
-    default_dimensions: Dict[str, str], optional
+    default_dimensions: dict[str, str], optional
         Metric dimensions as key=value that will always be present
 
 
@@ -624,7 +626,7 @@ def single_metric(
     SchemaValidationError
         When metric object fails EMF schema validation
     """  # noqa: E501
-    metric_set: Dict | None = None
+    metric_set: dict | None = None
     try:
         metric: SingleMetric = SingleMetric(namespace=namespace)
         metric.add_metric(name=name, unit=unit, value=value, resolution=resolution)

aws_lambda_powertools/metrics/functions.py

@@ -1,7 +1,6 @@
 from __future__ import annotations
 
 from datetime import datetime
-from typing import List
 
 from aws_lambda_powertools.metrics.provider.cloudwatch_emf.exceptions import (
     MetricResolutionError,
@@ -11,12 +10,12 @@
 from aws_lambda_powertools.shared import constants
 
 
-def extract_cloudwatch_metric_resolution_value(metric_resolutions: List, resolution: int | MetricResolution) -> int:
+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]
+    resolution : int | MetricResolution
         Metric resolution
 
     Returns
@@ -40,12 +39,12 @@ def extract_cloudwatch_metric_resolution_value(metric_resolutions: List, resolut
     )
 
 
-def extract_cloudwatch_metric_unit_value(metric_units: List, metric_valid_options: List, unit: str | MetricUnit) -> str:
+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]
+    unit : str | MetricUnit
         Metric unit
 
     Returns

aws_lambda_powertools/metrics/metrics.py

@@ -1,12 +1,14 @@
 # NOTE: keeps for compatibility
 from __future__ import annotations
 
-from typing import Any, Dict
+from typing import TYPE_CHECKING, Any
 
-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
-from aws_lambda_powertools.shared.types import AnyCallableT
+
+if TYPE_CHECKING:
+    from aws_lambda_powertools.metrics.base import MetricResolution, MetricUnit
+    from aws_lambda_powertools.metrics.provider.cloudwatch_emf.types import CloudWatchEMFOutput
+    from aws_lambda_powertools.shared.types import AnyCallableT
 
 
 class Metrics:
@@ -72,10 +74,10 @@ def lambda_handler():
     # and not get caught by accident with metrics data loss, or data deduplication
     # e.g., m1 and m2 add metric ProductCreated, however m1 has 'version' dimension  but m2 doesn't
     # Result: ProductCreated is created twice as we now have 2 different EMF blobs
-    _metrics: Dict[str, Any] = {}
-    _dimensions: Dict[str, str] = {}
-    _metadata: Dict[str, Any] = {}
-    _default_dimensions: Dict[str, Any] = {}
+    _metrics: dict[str, Any] = {}
+    _dimensions: dict[str, str] = {}
+    _metadata: dict[str, Any] = {}
+    _default_dimensions: dict[str, Any] = {}
 
     def __init__(
         self,
@@ -116,9 +118,9 @@ def add_dimension(self, name: str, value: str) -> None:
 
     def serialize_metric_set(
         self,
-        metrics: Dict | None = None,
-        dimensions: Dict | None = None,
-        metadata: Dict | None = None,
+        metrics: dict | None = None,
+        dimensions: dict | None = None,
+        metadata: dict | None = None,
     ) -> CloudWatchEMFOutput:
         return self.provider.serialize_metric_set(metrics=metrics, dimensions=dimensions, metadata=metadata)
 
@@ -146,7 +148,7 @@ def log_metrics(
         lambda_handler: AnyCallableT | None = None,
         capture_cold_start_metric: bool = False,
         raise_on_empty_metrics: bool = False,
-        default_dimensions: Dict[str, str] | None = None,
+        default_dimensions: dict[str, str] | None = None,
         **kwargs,
     ):
         return self.provider.log_metrics(
@@ -163,7 +165,7 @@ def set_default_dimensions(self, **dimensions) -> None:
 
         Parameters
         ----------
-        dimensions : Dict[str, Any], optional
+        dimensions : dict[str, Any], optional
             metric dimensions as key=value
 
         Example

aws_lambda_powertools/metrics/provider/base.py

@@ -3,11 +3,13 @@
 import functools
 import logging
 from abc import ABC, abstractmethod
-from typing import Any
+from typing import TYPE_CHECKING, Any
 
 from aws_lambda_powertools.metrics.provider import cold_start
-from aws_lambda_powertools.shared.types import AnyCallableT
-from aws_lambda_powertools.utilities.typing import LambdaContext
+
+if TYPE_CHECKING:
+    from aws_lambda_powertools.shared.types import AnyCallableT
+    from aws_lambda_powertools.utilities.typing import LambdaContext
 
 logger = logging.getLogger(__name__)
 
@@ -40,7 +42,7 @@ def add_metric(self, *args: Any, **kwargs: Any) -> Any:
 
         Returns
         ----------
-        Dict
+        dict
             A combined metrics dictionary.
 
         Raises
@@ -66,7 +68,7 @@ def serialize_metric_set(self, *args: Any, **kwargs: Any) -> Any:
 
         Returns
         ----------
-        Dict
+        dict
             Serialized metrics
 
         Raises
@@ -172,7 +174,7 @@ 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
+        default_dimensions: dict[str, str], optional
             metric dimensions as key=value that will always be present
 
         Raises