Microsoft Azure Monitor Client Library for Python
—
Quality
Pending
Does it follow best practices?
Impact
Pending
No eval scenarios have been run
Comprehensive metric data retrieval and analysis capabilities for Azure resources, including metric values, definitions, namespaces, and baseline comparisons for performance monitoring and trend analysis.
Query metric values for Azure resources with flexible filtering, aggregation, and time range options.
def list(resource_uri: str, timespan: Optional[str] = None, interval: Optional[str] = None, metricnames: Optional[str] = None, aggregation: Optional[str] = None, top: Optional[int] = None, orderby: Optional[str] = None, filter: Optional[str] = None, result_type: Optional[ResultType] = None, metricnamespace: Optional[str] = None, auto_adjust_timegrain: Optional[bool] = None, validate_dimensions: Optional[bool] = None, rollupby: Optional[str] = None, **kwargs: Any) -> Response:
"""
Lists the metric values for a resource.
Parameters:
- resource_uri: str - Resource identifier
- timespan: Optional[str] - Time range (ISO 8601 format)
- interval: Optional[str] - Metric sampling interval
- metricnames: Optional[str] - Comma-separated list of metric names
- aggregation: Optional[str] - Aggregation types to retrieve
- top: Optional[int] - Maximum number of records to retrieve
- orderby: Optional[str] - Aggregation to order results by
- filter: Optional[str] - OData filter for dimensions
- result_type: Optional[ResultType] - Result granularity (Data or Metadata)
- metricnamespace: Optional[str] - Metric namespace to query
- auto_adjust_timegrain: Optional[bool] - Auto-adjust time grain
- validate_dimensions: Optional[bool] - Validate dimension names and values
- rollupby: Optional[str] - Dimension to aggregate results by
Returns:
Response - Metric data response
"""
def list_at_subscription_scope(region: str, timespan: Optional[str] = None, interval: Optional[str] = None, metricnames: Optional[str] = None, aggregation: Optional[str] = None, top: Optional[int] = None, orderby: Optional[str] = None, filter: Optional[str] = None, result_type: Optional[MetricResultType] = None, metricnamespace: Optional[str] = None, auto_adjust_timegrain: Optional[bool] = None, validate_dimensions: Optional[bool] = None, rollupby: Optional[str] = None, **kwargs: Any) -> Response:
"""
Lists metrics at subscription scope for a region.
Parameters:
- region: str - Azure region name
- timespan: Optional[str] - Time range (ISO 8601 format)
- interval: Optional[str] - Metric sampling interval
- metricnames: Optional[str] - Comma-separated list of metric names
- aggregation: Optional[str] - Aggregation types to retrieve
- top: Optional[int] - Maximum number of records to retrieve
- orderby: Optional[str] - Aggregation to order results by
- filter: Optional[str] - OData filter for dimensions
- result_type: Optional[MetricResultType] - Result granularity
- metricnamespace: Optional[str] - Metric namespace to query
- auto_adjust_timegrain: Optional[bool] - Auto-adjust time grain
- validate_dimensions: Optional[bool] - Validate dimension names and values
- rollupby: Optional[str] - Dimension to aggregate results by
Returns:
Response - Subscription-level metric data response
"""Retrieve available metric definitions and metadata for resources and subscriptions.
def list(resource_uri: str, metricnamespace: Optional[str] = None, **kwargs: Any) -> ItemPaged[MetricDefinition]:
"""
Lists the metric definitions for the resource.
Parameters:
- resource_uri: str - Resource identifier
- metricnamespace: Optional[str] - Metric namespace to filter by
Returns:
ItemPaged[MetricDefinition] - Available metric definitions
"""
def list_at_subscription_scope(region: str, metricnamespace: Optional[str] = None, **kwargs: Any) -> ItemPaged[SubscriptionScopeMetricDefinition]:
"""
Lists metric definitions at subscription scope for a region.
Parameters:
- region: str - Azure region name
- metricnamespace: Optional[str] - Metric namespace to filter by
Returns:
ItemPaged[SubscriptionScopeMetricDefinition] - Subscription-level metric definitions
"""Retrieve available metric namespaces for resources to understand metric categorization.
def list(resource_uri: str, start_time: Optional[str] = None, **kwargs: Any) -> ItemPaged[MetricNamespace]:
"""
Lists the metric namespaces for the resource.
Parameters:
- resource_uri: str - Resource identifier
- start_time: Optional[str] - Start time for namespace availability
Returns:
ItemPaged[MetricNamespace] - Available metric namespaces
"""Retrieve baseline metric values for performance comparison and anomaly detection.
def list(resource_uri: str, metricnames: Optional[str] = None, metricnamespace: Optional[str] = None, timespan: Optional[str] = None, interval: Optional[str] = None, aggregation: Optional[str] = None, sensitivities: Optional[str] = None, result_type: Optional[ResultType] = None, **kwargs: Any) -> ItemPaged[SingleMetricBaseline]:
"""
Lists the metric baselines for a resource.
Parameters:
- resource_uri: str - Resource identifier
- metricnames: Optional[str] - Comma-separated list of metric names
- metricnamespace: Optional[str] - Metric namespace
- timespan: Optional[str] - Time range for baselines
- interval: Optional[str] - Baseline sampling interval
- aggregation: Optional[str] - Aggregation types for baselines
- sensitivities: Optional[str] - Sensitivity levels for baselines
- result_type: Optional[ResultType] - Result granularity
Returns:
ItemPaged[SingleMetricBaseline] - Metric baseline data
"""from datetime import datetime, timedelta
# Get CPU metrics for the last hour
end_time = datetime.utcnow()
start_time = end_time - timedelta(hours=1)
timespan = f"{start_time.isoformat()}Z/{end_time.isoformat()}Z"
resource_uri = f"/subscriptions/{subscription_id}/resourceGroups/web-rg/providers/Microsoft.Compute/virtualMachines/web-server"
metrics_response = client.metrics.list(
resource_uri=resource_uri,
timespan=timespan,
interval="PT5M", # 5-minute intervals
metricnames="Percentage CPU,Network In Total,Network Out Total",
aggregation="Average,Maximum"
)
# Parse the response
metrics_data = metrics_response.value
for metric in metrics_data:
print(f"Metric: {metric.name.value}")
for timeseries in metric.timeseries:
for data_point in timeseries.data:
print(f" Time: {data_point.time_stamp}, Average: {data_point.average}, Maximum: {data_point.maximum}")# Get available metrics for a resource
definitions = client.metric_definitions.list(resource_uri=resource_uri)
for definition in definitions:
print(f"Metric: {definition.name.value}")
print(f" Unit: {definition.unit}")
print(f" Aggregations: {[agg.value for agg in definition.supported_aggregation_types]}")
if definition.dimensions:
print(f" Dimensions: {[dim.value for dim in definition.dimensions]}")
print()# Query subscription-level metrics for a region
region = "East US"
subscription_metrics = client.metrics.list_at_subscription_scope(
region=region,
timespan=timespan,
metricnames="Virtual Machine CPU Credits Consumed",
aggregation="Average"
)
subscription_data = subscription_metrics.value
for metric in subscription_data:
print(f"Subscription Metric: {metric.name.value}")
for timeseries in metric.timeseries:
for data_point in timeseries.data:
print(f" Time: {data_point.time_stamp}, Value: {data_point.average}")# Get baseline data for comparison
baselines = client.baselines.list(
resource_uri=resource_uri,
metricnames="Percentage CPU",
timespan=timespan,
interval="PT5M",
aggregation="Average",
sensitivities="Low,Medium,High"
)
for baseline in baselines:
print(f"Baseline for: {baseline.name}")
for timeseries in baseline.timeseries:
for data_point in timeseries.data:
print(f" Time: {data_point.time_stamp}")
print(f" Low Threshold: {data_point.low_threshold}")
print(f" High Threshold: {data_point.high_threshold}")# Query metrics with dimension filtering
filtered_metrics = client.metrics.list(
resource_uri=resource_uri,
timespan=timespan,
metricnames="Requests",
aggregation="Total",
filter="HttpStatus eq '200' or HttpStatus eq '404'"
)
for metric in filtered_metrics.value:
for timeseries in metric.timeseries:
# Each timeseries represents a unique dimension combination
dimensions = {dim.name: dim.value for dim in timeseries.metadatavalues}
print(f"Dimensions: {dimensions}")
for data_point in timeseries.data:
print(f" Requests: {data_point.total}")class Metrics:
"""Response containing metric data."""
cost: Optional[int] # Query cost
timespan: str # Time range of the data
interval: Optional[str] # Sampling interval
value: List[Metric] # List of metrics
namespace: Optional[str] # Metric namespace
resource_region: Optional[str] # Resource region
class Metric:
"""Individual metric data."""
id: str # Metric identifier
type: str # Metric type
name: LocalizableString # Metric name
display_description: Optional[str] # Metric description
error_code: Optional[str] # Error code if applicable
error_message: Optional[str] # Error message if applicable
unit: MetricUnit # Metric unit
timeseries: List[TimeSeriesElement] # Time series data
class TimeSeriesElement:
"""Time series data for a metric."""
metadatavalues: Optional[List[MetadataValue]] # Dimension values
data: List[MetricValue] # Data points
class MetricValue:
"""Individual metric data point."""
time_stamp: datetime # Data point timestamp
average: Optional[float] # Average value
minimum: Optional[float] # Minimum value
maximum: Optional[float] # Maximum value
total: Optional[float] # Total value
count: Optional[int] # Count value
class MetricDefinition:
"""Metric definition and metadata."""
is_dimension_required: Optional[bool] # Whether dimensions are required
resource_id: Optional[str] # Resource ID
namespace: Optional[str] # Metric namespace
name: Optional[LocalizableString] # Metric name
display_description: Optional[str] # Metric description
category: Optional[str] # Metric category
metric_class: Optional[MetricClass] # Metric classification
unit: Optional[MetricUnit] # Metric unit
primary_aggregation_type: Optional[AggregationType] # Primary aggregation
supported_aggregation_types: Optional[List[AggregationType]] # Supported aggregations
metric_availabilities: Optional[List[MetricAvailability]] # Availability information
id: Optional[str] # Definition ID
dimensions: Optional[List[LocalizableString]] # Available dimensions
class MetricNamespace:
"""Metric namespace information."""
id: Optional[str] # Namespace ID
type: Optional[str] # Namespace type
name: Optional[str] # Namespace name
classification: Optional[NamespaceClassification] # Namespace classification
properties: Optional[MetricNamespaceName] # Namespace properties
class SingleMetricBaseline:
"""Baseline data for a single metric."""
id: str # Baseline ID
type: str # Baseline type
name: str # Baseline name
properties: Optional[BaselineResponse] # Baseline properties
class BaselineResponse:
"""Baseline response data."""
timespan: Optional[str] # Time range
interval: Optional[str] # Sampling interval
aggregation: Optional[str] # Aggregation type
timestamps: Optional[List[datetime]] # Timestamps
baseline: Optional[List[Baseline]] # Baseline values
class Baseline:
"""Baseline threshold values."""
sensitivity: Optional[BaselineSensitivity] # Sensitivity level
low_thresholds: List[float] # Low threshold values
high_thresholds: List[float] # High threshold values
class MetadataValue:
"""Metric dimension metadata."""
name: LocalizableString # Dimension name
value: Optional[str] # Dimension value
MetricUnit = Union[
"Count", "Bytes", "Seconds", "CountPerSecond", "BytesPerSecond",
"Percent", "MilliSeconds", "ByteSeconds", "Unspecified", "Cores",
"MilliCores", "NanoCores", "BitsPerSecond"
]
MetricClass = Union["Availability", "Transactions", "Errors", "Latency", "Saturation"]
BaselineSensitivity = Union["Low", "Medium", "High"]
ResultType = Union["Data", "Metadata"]
MetricResultType = Union["Data", "Metadata"]Install with Tessl CLI
npx tessl i tessl/pypi-azure-mgmt-monitor