Metrics¶

Metrics collection utilities for Lambda handlers.

Metrics utilities for AWS Lambda handlers.

Provides utilities for collecting and publishing CloudWatch metrics using AWS Lambda Powertools.

EnhancedMetrics ¶

Bases: Metrics

Extended Metrics class with convenience methods.

Provides additional helper methods for common metric patterns like counting, duration tracking, and success/failure recording.

add_count_metric ¶

add_count_metric(name, value)

Add a count metric.

Parameters:

Name	Type	Description	Default
`name`	`str`	The metric name.	required
`value`	`float`	The count value.	required

Source code in src/aibs_informatics_aws_lambda/common/metrics.py

def add_count_metric(self, name: str, value: float):
    """Add a count metric.

    Args:
        name (str): The metric name.
        value (float): The count value.
    """
    self.add_metric(name=name, unit=MetricUnit.Count, value=value)

add_duration_metric ¶

add_duration_metric(start=None, end=None, name='')

Add a duration metric.

Parameters:

Name	Type	Description	Default
`start`	`Optional[datetime]`	The start timestamp. Defaults to module load time.	`None`
`end`	`Optional[datetime]`	The end timestamp. Defaults to current time.	`None`
`name`	`str`	Prefix for the metric name.	`''`

Source code in src/aibs_informatics_aws_lambda/common/metrics.py

def add_duration_metric(
    self, start: Optional[datetime] = None, end: Optional[datetime] = None, name: str = ""
):
    """Add a duration metric.

    Args:
        start (Optional[datetime]): The start timestamp. Defaults to module load time.
        end (Optional[datetime]): The end timestamp. Defaults to current time.
        name (str): Prefix for the metric name.
    """
    add_duration_metric(start=start, end=end, name=name, metrics=self)

add_failure_metric ¶

add_failure_metric(name='')

Record a failed operation metric.

Parameters:

Name	Type	Description	Default
`name`	`str`	Prefix for the metric names.	`''`

Source code in src/aibs_informatics_aws_lambda/common/metrics.py

def add_failure_metric(self, name: str = ""):
    """Record a failed operation metric.

    Args:
        name (str): Prefix for the metric names.
    """
    add_failure_metric(name=name, metrics=self)

add_success_metric ¶

add_success_metric(name='')

Record a successful operation metric.

Parameters:

Name	Type	Description	Default
`name`	`str`	Prefix for the metric names.	`''`

Source code in src/aibs_informatics_aws_lambda/common/metrics.py

def add_success_metric(self, name: str = ""):
    """Record a successful operation metric.

    Args:
        name (str): Prefix for the metric names.
    """
    add_success_metric(name=name, metrics=self)

MetricsMixins ¶

Bases: HandlerMixins

Mixin class providing CloudWatch metrics capabilities.

Integrates AWS Lambda Powertools Metrics for automatic metric collection and publishing to CloudWatch.

metrics `property` `writable` ¶

metrics

Get the metrics collector, creating one if needed.

Returns:

Type	Description
`EnhancedMetrics`	The EnhancedMetrics instance for this handler.

add_metric `classmethod` ¶

add_metric(metrics, name, value, unit=MetricUnit.Count)

Add a metric to the collector.

Parameters:

Name	Type	Description	Default
`metrics`	`Metrics`	The metrics collector.	required
`name`	`str`	The metric name.	required
`value`	`float`	The metric value.	required
`unit`	`MetricUnit`	The metric unit. Defaults to Count.	`Count`

Source code in src/aibs_informatics_aws_lambda/common/metrics.py

@classmethod
def add_metric(
    cls,
    metrics: Metrics,
    name: str,
    value: float,
    unit: MetricUnit = MetricUnit.Count,
):
    """Add a metric to the collector.

    Args:
        metrics (Metrics): The metrics collector.
        name (str): The metric name.
        value (float): The metric value.
        unit (MetricUnit): The metric unit. Defaults to Count.
    """
    metrics.add_metric(name=name, unit=unit, value=value)

get_metrics `classmethod` ¶

get_metrics(
    service=None, namespace=None, **additional_dimensions
)

Create a new EnhancedMetrics instance.

Parameters:

Name	Type	Description	Default
`service`	`Optional[str]`	The service name for metrics.	`None`
`namespace`	`Optional[str]`	The CloudWatch namespace.	`None`
`**additional_dimensions`	`str`	Additional metric dimensions as key-value pairs.	`{}`

Returns:

Type	Description
`EnhancedMetrics`	A configured EnhancedMetrics instance.

Source code in src/aibs_informatics_aws_lambda/common/metrics.py

@classmethod
def get_metrics(
    cls,
    service: Optional[str] = None,
    namespace: Optional[str] = None,
    **additional_dimensions: str,
) -> EnhancedMetrics:
    """Create a new EnhancedMetrics instance.

    Args:
        service (Optional[str]): The service name for metrics.
        namespace (Optional[str]): The CloudWatch namespace.
        **additional_dimensions (str): Additional metric dimensions as key-value pairs.

    Returns:
        A configured EnhancedMetrics instance.
    """
    metrics = EnhancedMetrics(service=service, namespace=namespace)
    for dimension_name, dimension_value in additional_dimensions.items():
        metrics.add_dimension(name=dimension_name, value=dimension_value)
    return metrics

add_duration_metric ¶

add_duration_metric(
    start=None, end=None, name="", metrics=None
)

Add a duration metric to the metrics collector.

Calculates the duration between start and end times and records it as a CloudWatch metric in milliseconds.

Parameters:

Name	Type	Description	Default
`start`	`Optional[datetime]`	The start timestamp. Defaults to module load time.	`None`
`end`	`Optional[datetime]`	The end timestamp. Defaults to current time.	`None`
`name`	`str`	Prefix for the metric name. Final name is '{name}Duration'.	`''`
`metrics`	`Optional[Union[EphemeralMetrics, Metrics]]`	The metrics collector to use. Creates ephemeral if None.	`None`

Source code in src/aibs_informatics_aws_lambda/common/metrics.py

def add_duration_metric(
    start: Optional[datetime] = None,
    end: Optional[datetime] = None,
    name: str = "",
    metrics: Optional[Union[EphemeralMetrics, Metrics]] = None,
):
    """Add a duration metric to the metrics collector.

    Calculates the duration between start and end times and records
    it as a CloudWatch metric in milliseconds.

    Args:
        start (Optional[datetime]): The start timestamp. Defaults to module load time.
        end (Optional[datetime]): The end timestamp. Defaults to current time.
        name (str): Prefix for the metric name. Final name is '{name}Duration'.
        metrics (Optional[Union[EphemeralMetrics, Metrics]]): The metrics collector to use.
            Creates ephemeral if None.
    """
    start = start or DEFAULT_TIME_START
    end = end or datetime.now(start.tzinfo)
    duration = end - start
    if metrics is None:
        metrics = EphemeralMetrics()
    metrics.add_metric(
        name=f"{name}Duration", unit=MetricUnit.Milliseconds, value=duration.total_seconds() * 1000
    )

add_failure_metric ¶

add_failure_metric(name='', metrics=None)

Record a failed operation metric.

Adds metrics indicating success (0) and failure (1) counts.

Parameters:

Name	Type	Description	Default
`name`	`str`	Prefix for the metric names.	`''`
`metrics`	`Optional[Union[EphemeralMetrics, Metrics]]`	The metrics collector to use. Creates ephemeral if None.	`None`

Source code in src/aibs_informatics_aws_lambda/common/metrics.py

def add_failure_metric(name: str = "", metrics: Optional[Union[EphemeralMetrics, Metrics]] = None):
    """Record a failed operation metric.

    Adds metrics indicating success (0) and failure (1) counts.

    Args:
        name (str): Prefix for the metric names.
        metrics (Optional[Union[EphemeralMetrics, Metrics]]): The metrics collector to use.
            Creates ephemeral if None.
    """
    if metrics is None:
        metrics = EphemeralMetrics()
    metrics.add_metric(name=f"{name}Success", unit=MetricUnit.Count, value=0)
    metrics.add_metric(name=f"{name}Failure", unit=MetricUnit.Count, value=1)

add_success_metric ¶

add_success_metric(name='', metrics=None)

Record a successful operation metric.

Adds metrics indicating success (1) and failure (0) counts.

Parameters:

Name	Type	Description	Default
`name`	`str`	Prefix for the metric names.	`''`
`metrics`	`Optional[Union[EphemeralMetrics, Metrics]]`	The metrics collector to use. Creates ephemeral if None.	`None`

Source code in src/aibs_informatics_aws_lambda/common/metrics.py

def add_success_metric(name: str = "", metrics: Optional[Union[EphemeralMetrics, Metrics]] = None):
    """Record a successful operation metric.

    Adds metrics indicating success (1) and failure (0) counts.

    Args:
        name (str): Prefix for the metric names.
        metrics (Optional[Union[EphemeralMetrics, Metrics]]): The metrics collector to use.
            Creates ephemeral if None.
    """
    if metrics is None:
        metrics = EphemeralMetrics()
    metrics.add_metric(name=f"{name}Success", unit=MetricUnit.Count, value=1)
    metrics.add_metric(name=f"{name}Failure", unit=MetricUnit.Count, value=0)

Metrics¶

EnhancedMetrics ¶

add_count_metric ¶

add_duration_metric ¶

add_failure_metric ¶

add_success_metric ¶

MetricsMixins ¶

metrics property writable ¶

add_metric classmethod ¶

get_metrics classmethod ¶

add_duration_metric ¶

add_failure_metric ¶

add_success_metric ¶

metrics `property` `writable` ¶

add_metric `classmethod` ¶

get_metrics `classmethod` ¶