Skip to content

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

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged
merged 1 commit into from
Aug 15, 2024
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
58 changes: 30 additions & 28 deletions aws_lambda_powertools/metrics/base.py
Original file line number Diff line number Diff line change
Expand Up @@ -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,
Expand All @@ -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
Expand Down Expand Up @@ -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 {}
Expand Down Expand Up @@ -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
Expand All @@ -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))
Expand All @@ -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
Expand All @@ -172,7 +174,7 @@ def serialize_metric_set(

Returns
-------
Dict
dict
Serialized metrics following EMF specification

Raises
Expand Down Expand Up @@ -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]
Expand Down Expand Up @@ -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.

Expand Down Expand Up @@ -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
Expand Down Expand Up @@ -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
Expand All @@ -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
Expand Down Expand Up @@ -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

Expand Down Expand Up @@ -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


Expand All @@ -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)
Expand Down
9 changes: 4 additions & 5 deletions aws_lambda_powertools/metrics/functions.py
Original file line number Diff line number Diff line change
@@ -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,
Expand All @@ -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
Expand All @@ -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
Expand Down
28 changes: 15 additions & 13 deletions aws_lambda_powertools/metrics/metrics.py
Original file line number Diff line number Diff line change
@@ -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:
Expand Down Expand Up @@ -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,
Expand Down Expand Up @@ -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)

Expand Down Expand Up @@ -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(
Expand All @@ -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
Expand Down
14 changes: 8 additions & 6 deletions aws_lambda_powertools/metrics/provider/base.py
Original file line number Diff line number Diff line change
Expand Up @@ -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__)

Expand Down Expand Up @@ -40,7 +42,7 @@ def add_metric(self, *args: Any, **kwargs: Any) -> Any:

Returns
----------
Dict
dict
A combined metrics dictionary.

Raises
Expand All @@ -66,7 +68,7 @@ def serialize_metric_set(self, *args: Any, **kwargs: Any) -> Any:

Returns
----------
Dict
dict
Serialized metrics

Raises
Expand Down Expand Up @@ -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
Expand Down
Loading
Loading