Azure Monitor Query Py

Azure Monitor Query SDK for Python

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

Installation

pip install azure-monitor-query

Environment Variables

# Log Analytics
AZURE_LOG_ANALYTICS_WORKSPACE_ID=  # Required for log queries

# Metrics
AZURE_METRICS_RESOURCE_URI=/subscriptions//resourceGroups//providers///  # Required for metric queries
AZURE_TOKEN_CREDENTIALS=prod # Required only if DefaultAzureCredential is used in production

Authentication & Lifecycle

> 🔑 Two rules apply to every code sample below: > > 1. Prefer DefaultAzureCredential. It works locally (Azure CLI / VS Code / Developer CLI) and in Azure (managed identity, workload identity) with no code change. Avoid connection strings, account/API keys — they bypass Entra audit and rotation. > - Local dev: DefaultAzureCredential works as-is. > - Production: set AZURE_TOKEN_CREDENTIALS=prod (or AZURE_TOKEN_CREDENTIALS=) to constrain the credential chain to production-safe credentials. > 2. Wrap every client in a context manager so HTTP transports, sockets, and token caches are released deterministically: > - Sync: with (...) as client: > - Async: async with (...) as client: and async with DefaultAzureCredential() as credential: (from azure.identity.aio) > > Snippets may abbreviate this setup, but production code should always follow both rules.

from azure.identity import DefaultAzureCredential, ManagedIdentityCredential

# Local dev: DefaultAzureCredential. Production: set AZURE_TOKEN_CREDENTIALS=prod or AZURE_TOKEN_CREDENTIALS=
credential = DefaultAzureCredential(require_envvar=True)
# Or use a specific credential directly in production:
# See https://learn.microsoft.com/python/api/overview/azure/identity-readme?view=azure-python#credential-classes
# credential = ManagedIdentityCredential()

Logs Query Client

Basic Query

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

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

with LogsQueryClient(credential) as client:
    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 Query Client

Query Resource Metrics

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

with MetricsQueryClient(credential) as metrics_client:
    response = metrics_client.query_resource(
        resource_uri=os.environ["AZURE_METRICS_RESOURCE_URI"],
        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.query import MetricAggregationType

response = metrics_client.query_resource(
    resource_uri=resource_uri,
    metric_names=["Requests"],
    timespan=timedelta(hours=1),
    aggregations=[
        MetricAggregationType.AVERAGE,
        MetricAggregationType.MAXIMUM,
        MetricAggregationType.MINIMUM,
        MetricAggregationType.COUNT
    ]
)

Filter by Dimension

response = metrics_client.query_resource(
    resource_uri=resource_uri,
    metric_names=["Requests"],
    timespan=timedelta(hours=1),
    filter="ApiName eq 'GetBlob'"
)

List Metric Definitions

definitions = metrics_client.list_metric_definitions(resource_uri)
for definition in definitions:
    print(f"{definition.name}: {definition.unit}")

List Metric Namespaces

namespaces = metrics_client.list_metric_namespaces(resource_uri)
for ns in namespaces:
    print(ns.fully_qualified_namespace)

Async Clients

from azure.monitor.query.aio import LogsQueryClient, MetricsQueryClient
from azure.identity.aio import DefaultAzureCredential

async def query_logs():
    async with DefaultAzureCredential() as credential:
        async with LogsQueryClient(credential) as client:
            response = await client.query_workspace(
                workspace_id=workspace_id,
                query="AppRequests | take 10",
                timespan=timedelta(hours=1)
            )
            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 | | MetricsQueryClient | Query Azure Monitor metrics |

Best Practices

1. Pick sync OR async and stay consistent. Do not mix azure.xxx sync clients with azure.xxx.aio async clients in the same call path. Choose one mode per module.
2. Always use context managers for clients and async credentials. Wrap every client in with Client(...) as client: (sync) or async with Client(...) as client: (async). For async DefaultAzureCredential from azure.identity.aio, also use async with credential: so tokens and transports are cleaned up.
3. Use DefaultAzureCredential for portable auth across local dev and Azure (avoid connection strings / API keys when possible).
4. Use timedelta for relative time ranges
5. Handle partial results for large queries
6. Use batch queries when running multiple queries
7. Set appropriate granularity for metrics to reduce data points
8. Convert to DataFrame for easier data analysis
9. Use aggregations to summarize metric data
10. Filter by dimensions to narrow metric results

Source & license

This open-source skill is cataloged on AgentStack and links to its original source — we do not rehost the code.

• Author: simonpsson
• Source: simonpsson/claude-skills
• License: MIT

Install and usage instructions live in the source repository linked above.