azure-monitor-query-py

name	azure-monitor-query-py
description	Azure Monitor Query SDK for Python. Use for querying Log Analytics workspaces and Azure Monitor metrics.
risk	unknown
source	community
date_added	2026-02-27

Azure Monitor Query SDK for Python

Query logs and metrics from Azure Monitor and Log Analytics workspaces.

Installation

pip install azure-monitor-query
pip install azure-monitor-querymetrics

Environment Variables

# Log Analytics
AZURE_LOG_ANALYTICS_WORKSPACE_ID=<workspace-id>

# Metrics
AZURE_METRICS_RESOURCE_URI=/subscriptions/<sub>/resourceGroups/<rg>/providers/<provider>/<type>/<name>

Authentication

from azure.identity import DefaultAzureCredential

credential = DefaultAzureCredential()

Logs Query Client

Basic Query

from azure.monitor.query import LogsQueryClient
from datetime import timedelta

client = LogsQueryClient(credential)

query = """
AppRequests
| where TimeGenerated > ago(1h)
| summarize count() by bin(TimeGenerated, 5m), ResultCode
| order by TimeGenerated desc
"""

response = client.query_workspace(
    workspace_id=os.environ["AZURE_LOG_ANALYTICS_WORKSPACE_ID"],
    query=query,
    timespan=timedelta(hours=1)
)

for table in response.tables:
    for row in table.rows:
        print(row)

Query with Time Range

from datetime import datetime, timezone

response = client.query_workspace(
    workspace_id=workspace_id,
    query="AppRequests | take 10",
    timespan=(
        datetime(2024, 1, 1, tzinfo=timezone.utc),
        datetime(2024, 1, 2, tzinfo=timezone.utc)
    )
)

Convert to DataFrame

import pandas as pd

response = client.query_workspace(workspace_id, query, timespan=timedelta(hours=1))

if response.tables:
    table = response.tables[0]
    df = pd.DataFrame(data=table.rows, columns=[col.name for col in table.columns])
    print(df.head())

Batch Query

from azure.monitor.query import LogsBatchQuery

queries = [
    LogsBatchQuery(workspace_id=workspace_id, query="AppRequests | take 5", timespan=timedelta(hours=1)),
    LogsBatchQuery(workspace_id=workspace_id, query="AppExceptions | take 5", timespan=timedelta(hours=1))
]

responses = client.query_batch(queries)

for response in responses:
    if response.tables:
        print(f"Rows: {len(response.tables[0].rows)}")

Handle Partial Results

from azure.monitor.query import LogsQueryStatus

response = client.query_workspace(workspace_id, query, timespan=timedelta(hours=24))

if response.status == LogsQueryStatus.PARTIAL:
    print(f"Partial results: {response.partial_error}")
elif response.status == LogsQueryStatus.FAILURE:
    print(f"Query failed: {response.partial_error}")

Metrics Client

Query Resource Metrics

from azure.monitor.querymetrics import MetricsClient
from datetime import timedelta

metrics_client = MetricsClient(credential)

responses = metrics_client.query_resources(
    resource_ids=[os.environ["AZURE_METRICS_RESOURCE_URI"]],
    metric_namespace="Microsoft.Storage/storageAccounts", # Example namespace
    metric_names=["Percentage CPU", "Network In Total"],
    timespan=timedelta(hours=1),
    granularity=timedelta(minutes=5)
)

for response in responses:
    for metric in response.metrics:
    metric_names=["Percentage CPU", "Network In Total"],
    timespan=timedelta(hours=1),
    granularity=timedelta(minutes=5)
)

for metric in response.metrics:
    print(f"{metric.name}:")
    for time_series in metric.timeseries:
        for data in time_series.data:
            print(f"  {data.timestamp}: {data.average}")

Aggregations

from azure.monitor.querymetrics import MetricAggregationType

response = metrics_client.query_resources(
    resource_ids=[resource_uri],
    metric_namespace="Microsoft.Storage/storageAccounts", # Example namespace
    metric_names=["Requests"],
    timespan=timedelta(hours=1),
    aggregations=[
        MetricAggregationType.AVERAGE,
        MetricAggregationType.MAXIMUM,
        MetricAggregationType.MINIMUM,
        MetricAggregationType.COUNT
    ]
)

Filter by Dimension

response = metrics_client.query_resources(
    resource_ids=[resource_uri],
    metric_namespace="Microsoft.Storage/storageAccounts", # Example namespace
    metric_names=["Requests"],
    timespan=timedelta(hours=1),
    filter="ApiName eq 'GetBlob'"
)

Async Clients

from azure.monitor.query.aio import LogsQueryClient
from azure.monitor.querymetrics.aio import MetricsClient
from azure.identity.aio import DefaultAzureCredential

async def query_logs():
    credential = DefaultAzureCredential()
    client = LogsQueryClient(credential)
    
    response = await client.query_workspace(
        workspace_id=workspace_id,
        query="AppRequests | take 10",
        timespan=timedelta(hours=1)
    )
    
    await client.close()
    await credential.close()
    return response

Common Kusto Queries

// Requests by status code
AppRequests
| summarize count() by ResultCode
| order by count_ desc

// Exceptions over time
AppExceptions
| summarize count() by bin(TimeGenerated, 1h)

// Slow requests
AppRequests
| where DurationMs > 1000
| project TimeGenerated, Name, DurationMs
| order by DurationMs desc

// Top errors
AppExceptions
| summarize count() by ExceptionType
| top 10 by count_

Client Types

Client	Purpose
LogsQueryClient	Query Log Analytics workspaces (azure-monitor-query)
MetricsClient	Query Azure Monitor metrics (azure-monitor-querymetrics)

Best Practices

1. Use timedelta for relative time ranges
2. Handle partial results for large queries
3. Use batch queries when running multiple queries
4. Set appropriate granularity for metrics to reduce data points
5. Convert to DataFrame for easier data analysis
6. Use aggregations to summarize metric data
7. Filter by dimensions to narrow metric results

When to Use

This skill is applicable to execute the workflow or actions described in the overview.