tessl/pypi-azure

Microsoft Azure Client Libraries for Python meta-package providing comprehensive access to Azure cloud services and management capabilities

—

Pending

Quality

Pending

Does it follow best practices?

Impact

Pending

No eval scenarios have been run

Overview

Eval results

Files

Monitoring and Analytics

Name: tessl/pypi-azure
Author: tessl

Azure monitoring and analytics services provide application performance monitoring, log analytics, and operational insights. This includes Application Insights for application performance monitoring and Log Analytics for log data analysis and visualization.

Capabilities

Azure Application Insights

Provides application performance monitoring and analytics for web applications and services. Collects telemetry data including requests, dependencies, exceptions, and custom metrics.

class ApplicationInsightsDataClient:
    """
    Client for Application Insights Data API operations.
    
    Parameters:
    - credentials: Authentication credentials
    """
    def __init__(self, credentials, **kwargs): ...
    
    @property
    def query(self): ...  # Query operations
    
    @property
    def metrics(self): ...  # Metrics operations
    
    @property
    def events(self): ...  # Events operations

Query Operations

Execute KQL (Kusto Query Language) queries against Application Insights data.

class QueryOperations:
    def execute(self, app_id: str, body, **kwargs):
        """
        Execute a KQL query against Application Insights data.
        
        Parameters:
        - app_id: str, Application Insights application ID
        - body: QueryBody, Query request body containing KQL query
        
        Returns:
        QueryResults object with query results
        """

Metrics Operations

Retrieve pre-aggregated metrics from Application Insights applications.

class MetricsOperations:
    def get(self, app_id: str, metric_id: str, **kwargs):
        """
        Get a specific metric from an Application Insights application.
        
        Parameters:
        - app_id: str, Application Insights application ID
        - metric_id: str, Metric identifier
        
        Returns:
        MetricsResult object with metric data
        """
    
    def get_multiple(self, app_id: str, body, **kwargs):
        """
        Get multiple metrics from an Application Insights application.
        
        Parameters:
        - app_id: str, Application Insights application ID
        - body: MetricsPostBody, Request body with metric specifications
        
        Returns:
        List of MetricsResult objects
        """

Events Operations

Retrieve event data from Application Insights applications.

class EventsOperations:
    def get_by_type(self, app_id: str, event_type: str, event_id: str, **kwargs):
        """
        Get a specific event by type and ID.
        
        Parameters:
        - app_id: str, Application Insights application ID
        - event_type: str, Event type (requests, dependencies, exceptions, etc.)
        - event_id: str, Event identifier
        
        Returns:
        EventsResult object with event data
        """
    
    def get(self, app_id: str, event_type: str, **kwargs):
        """
        Get events of a specific type.
        
        Parameters:
        - app_id: str, Application Insights application ID  
        - event_type: str, Event type to retrieve
        
        Returns:
        EventsResults object with events data
        """

Azure Log Analytics

Provides log data collection, analysis, and visualization capabilities. Supports KQL queries for advanced log analysis and custom dashboards.

class LogAnalyticsDataClient:
    """
    Client for Azure Log Analytics Data API operations.
    
    Parameters:
    - credentials: Authentication credentials
    """
    def __init__(self, credentials, **kwargs): ...
    
    @property  
    def query(self): ...  # Query operations

Log Analytics Query Operations

Execute KQL queries against Log Analytics workspaces.

class QueryOperations:
    def execute(self, workspace_id: str, body, **kwargs):
        """
        Execute a KQL query against a Log Analytics workspace.
        
        Parameters:
        - workspace_id: str, Log Analytics workspace ID
        - body: QueryBody, Query request body with KQL query
        
        Returns:
        QueryResults object with query results
        """

Data Models

Application Insights Models

class QueryBody:
    """Request body for Application Insights queries."""
    def __init__(self, query: str, **kwargs): ...
    
    query: str  # KQL query string
    timespan: str  # Time range for the query (optional, ISO 8601 duration)
    applications: list  # Additional application IDs to include (optional)

class QueryResults:
    """Results from an Application Insights query."""
    def __init__(self): ...
    
    tables: list  # Result tables
    error: object  # Error information if query failed

class QueryResultsTable:
    """Individual table from query results.""" 
    def __init__(self): ...
    
    name: str  # Table name
    columns: list  # Column definitions
    rows: list  # Data rows

class Column:
    """Column definition in query results."""
    def __init__(self): ...
    
    name: str  # Column name
    type: str  # Column data type (string, datetime, long, real, etc.)

class MetricsPostBody:
    """Request body for retrieving multiple metrics."""
    def __init__(self, metrics: list, **kwargs): ...
    
    metrics: list  # List of metric specifications
    timespan: str  # Time range (optional)
    interval: str  # Aggregation interval (optional)
    aggregation: list  # Aggregation types (optional)
    segment: list  # Segmentation dimensions (optional)
    top: int  # Top N results (optional)
    orderby: str  # Sort order (optional)
    filter: str  # Filter expression (optional)

class MetricsResult:
    """Result from a metrics query."""
    def __init__(self): ...
    
    value: object  # Metric value data
    start: str  # Start time of the metric period
    end: str  # End time of the metric period

class EventsResults:
    """Results from an events query."""
    def __init__(self): ...
    
    value: list  # List of events
    ai_messages: list  # AI-generated insights (optional)

class EventsResult:
    """Individual event result."""
    def __init__(self): ...
    
    id: str  # Event identifier
    count: int  # Event occurrence count
    timestamp: str  # Event timestamp
    customDimensions: dict  # Custom properties
    customMeasurements: dict  # Custom measurements

Log Analytics Models

class QueryBody:
    """Request body for Log Analytics queries."""
    def __init__(self, query: str, **kwargs): ...
    
    query: str  # KQL query string
    timespan: str  # Time range for the query (optional)
    workspaces: list  # Additional workspace IDs to query (optional)

class QueryResults:
    """Results from a Log Analytics query."""
    def __init__(self): ...
    
    tables: list  # Result tables
    error: object  # Error information if query failed
    
class ErrorInfo:
    """Error information for failed queries."""
    def __init__(self): ...
    
    code: str  # Error code
    message: str  # Error message
    details: list  # Additional error details

Usage Examples

Querying Application Insights Data

from azure.applicationinsights import ApplicationInsightsDataClient
from azure.applicationinsights.models import QueryBody

# Create Application Insights client
ai_client = ApplicationInsightsDataClient(credentials)

# Application ID from Application Insights
app_id = "12345678-1234-1234-1234-123456789abc"

# Execute a KQL query
query_body = QueryBody(
    query="""
    requests
    | where timestamp > ago(24h)
    | summarize count() by bin(timestamp, 1h), resultCode
    | order by timestamp desc
    """,
    timespan="P1D"  # Last 1 day
)

results = ai_client.query.execute(app_id, query_body)

# Process results
for table in results.tables:
    print(f"Table: {table.name}")
    print("Columns:", [col.name for col in table.columns])
    
    for row in table.rows:
        print("Row:", row)

# Query for exceptions
exception_query = QueryBody(
    query="""
    exceptions
    | where timestamp > ago(1h)
    | project timestamp, type, outerMessage, operation_Name
    | order by timestamp desc
    | take 10
    """
)

exception_results = ai_client.query.execute(app_id, exception_query)
print(f"Found {len(exception_results.tables[0].rows)} exceptions in the last hour")

Retrieving Application Insights Metrics

from azure.applicationinsights.models import MetricsPostBody

# Get request rate metric
request_metric = ai_client.metrics.get(
    app_id=app_id,
    metric_id="requests/rate"
)

print(f"Request rate: {request_metric.value}")

# Get multiple metrics
metrics_body = MetricsPostBody(
    metrics=[
        {"id": "requests/rate"},
        {"id": "requests/duration"},
        {"id": "exceptions/rate"}
    ],
    timespan="PT1H",  # Last 1 hour
    interval="PT5M"   # 5-minute intervals
)

multiple_metrics = ai_client.metrics.get_multiple(app_id, metrics_body)

for metric in multiple_metrics:
    print(f"Metric: {metric.value}")

Working with Application Insights Events

# Get recent requests
requests = ai_client.events.get(
    app_id=app_id,
    event_type="requests"
)

print(f"Found {len(requests.value)} requests")

for request in requests.value[:5]:  # Show first 5
    print(f"Request: {request.id}")
    print(f"Timestamp: {request.timestamp}")
    print(f"Custom dimensions: {request.customDimensions}")

# Get a specific event
specific_event = ai_client.events.get_by_type(
    app_id=app_id,
    event_type="requests",
    event_id="some-request-id"
)

print(f"Event details: {specific_event.id}")

Querying Log Analytics Data

from azure.loganalytics import LogAnalyticsDataClient
from azure.loganalytics.models import QueryBody

# Create Log Analytics client
la_client = LogAnalyticsDataClient(credentials)

# Workspace ID from Log Analytics
workspace_id = "12345678-1234-1234-1234-123456789abc"

# Execute a KQL query
query_body = QueryBody(
    query="""
    Heartbeat
    | where TimeGenerated > ago(1h)
    | summarize count() by Computer, bin(TimeGenerated, 5m)
    | order by TimeGenerated desc
    """,
    timespan="PT1H"
)

results = la_client.query.execute(workspace_id, query_body)

# Process results
if results.tables:
    table = results.tables[0]
    print(f"Query returned {len(table.rows)} rows")
    
    # Print column headers
    headers = [col.name for col in table.columns]
    print("Columns:", headers)
    
    # Print first few rows
    for row in table.rows[:10]:
        print("Row:", dict(zip(headers, row)))

# Query for security events
security_query = QueryBody(
    query="""
    SecurityEvent
    | where TimeGenerated > ago(24h)
    | where EventID == 4625  // Failed logon attempts
    | summarize FailedAttempts = count() by Account, IpAddress
    | where FailedAttempts > 5
    | order by FailedAttempts desc
    """
)

security_results = la_client.query.execute(workspace_id, security_query)
print(f"Found {len(security_results.tables[0].rows)} accounts with multiple failed logons")

Error Handling

from azure.applicationinsights.models import QueryBody

try:
    # Execute query with syntax error
    bad_query = QueryBody(query="invalid KQL syntax")
    results = ai_client.query.execute(app_id, bad_query)
    
    if results.error:
        print(f"Query error: {results.error.message}")
        print(f"Error code: {results.error.code}")
        
except Exception as e:
    print(f"Request failed: {e}")

# Check for empty results
query_body = QueryBody(
    query="requests | where timestamp > ago(1d) | take 0"
)

results = ai_client.query.execute(app_id, query_body)

if not results.tables or not results.tables[0].rows:
    print("No data found for the specified time range")
else:
    print(f"Found {len(results.tables[0].rows)} results")

Advanced KQL Queries

# Performance analysis query
perf_query = QueryBody(
    query="""
    requests
    | where timestamp > ago(24h)
    | join (dependencies | where timestamp > ago(24h)) on operation_Id
    | project 
        timestamp,
        name,
        duration,
        dependency_duration = duration1,
        dependency_name = name1,
        success,
        resultCode
    | where duration > 1000  // Slow requests (>1 second)
    | order by duration desc
    | take 20
    """
)

perf_results = ai_client.query.execute(app_id, perf_query)

# User behavior analysis
user_query = QueryBody(
    query="""
    pageViews
    | where timestamp > ago(7d)
    | extend hour = bin(timestamp, 1h)
    | summarize 
        UniqueUsers = dcount(user_Id),
        PageViews = count(),
        AvgDuration = avg(duration)
    by hour
    | order by hour desc
    """
)

user_results = ai_client.query.execute(app_id, user_query)

Install with Tessl CLI