actvalue.azure-errors

actvalue.azure-errors

Structured JSON logging and error handling for Python Azure Functions (v2 programming model), designed for Application Insights consumption.

Features

• Structured logs — every log entry is a JSON object with FunctionApp, FunctionName, InvocationId, Severity, and more
• ContextVar-based context — no need to pass context through every function signature
• Error hierarchy — NotFoundError (404), ForbiddenError (403), BadRequestError (400) with automatic structured logging
• HTTP error handler — on_error() catches any error and returns a structured HttpResponse
• Queue error handler — on_queue_error() logs structured errors and re-raises for Azure runtime retry/dead-lettering
• Application Insights integration — Error and Warning severity logs automatically sent to the exceptions table via azure-monitor-opentelemetry

Install

pip install actvalue.azure-errors

Optional dependencies:

# Azure Functions context support
pip install actvalue.azure-errors[azure]

Quick Start

1. Set the app name and Application Insights exception tracker (once per function app)

from azure_errors import (
    configure_app_insights_exceptions,
    set_function_app,
    silence_azure_sdk_logs,
)

set_function_app('my-function-app')

# Stop Azure SDK / MSAL / urllib3 INFO chatter (HTTP request+response dumps,
# token acquisition, connection-pool noise) from reaching the App Insights
# `traces` table via the Functions worker's root-logger handler.
silence_azure_sdk_logs()

# Optional: enable writing Error/Warning logs to the App Insights `exceptions`
# table. Builds a standalone AzureMonitorLogExporter — we deliberately do NOT
# call `configure_azure_monitor()`, which would install auto-instrumentation
# for requests/urllib3/httpx/etc. and flood the `dependencies` / `requests`
# tables. Requires: pip install actvalue.azure-errors[appinsights]
import os
conn = os.environ.get('APPLICATIONINSIGHTS_CONNECTION_STRING')
if conn:
    configure_app_insights_exceptions(conn)

If you need a custom exporter or client, use the lower-level set_exception_tracker(fn) hook instead — fn receives (exception, properties) for every Error / Warning log.

2. Wrap HTTP handlers with run_with_context

import azure.functions as func
from azure_errors import run_with_context, on_error

async def my_handler(req: func.HttpRequest, context: func.Context) -> func.HttpResponse:
    try:
        return await run_with_context(context, lambda: _handle(req))
    except Exception as e:
        return on_error(context, e)

async def _handle(req: func.HttpRequest) -> func.HttpResponse:
    # your logic here — context is available anywhere via get_context()
    return func.HttpResponse("OK", status_code=200)

3. Wrap queue/timer handlers with run_with_context and on_queue_error

import azure.functions as func
from azure_errors import run_with_context, on_queue_error

def handler(msg: func.ServiceBusMessage, context: func.Context) -> None:
    try:
        run_with_context(context, lambda: process_message(msg))
    except Exception as e:
        on_queue_error(context, e)  # logs structured + re-raises

4. Log from anywhere in the call stack

from azure_errors import log_info, log_warning, log_error

# No need to pass context — resolved via ContextVar

# Pass a string code
log_info('Processing device', device_code)
log_warning('DuplicateKey', 'Some records not inserted', 'CODE1')
log_error('BlobStorageError', 'Write failed', mac)

# Or pass a details dict
log_info('Processing batch', {'Store_id': 1, 'Area_id': 2})
log_warning('DuplicateKey', 'Some records not inserted', {'table': 'devices', 'count': 3})
log_error('BlobStorageError', 'Write failed', {'container': 'images', 'blob': 'test.png'})

5. Raise typed errors

from azure_errors import NotFoundError, ForbiddenError, BadRequestError

raise NotFoundError('Device not found', device_code)
# → HTTP 404, severity: Warning, structured log emitted by on_error()

StructuredLog Shape

Every log entry emitted by this library follows this structure:

{
  "FunctionApp": "my-function-app",
  "FunctionName": "my_handler",
  "InvocationId": "abc-123",
  "ErrorType": "NotFoundError",
  "Severity": "Warning",
  "Message": "Device not found",
  "Code": "DEV001",
  "Details": {}
}

API

Context

Logging

Errors

Severity Routing

Logs are routed to the appropriate Python logging level:

When no context is available (e.g. startup code), logs still go through logging with FunctionName and InvocationId set to "unknown".

Application Insights Integration

By default, the standard Python logging calls emitted by this library are captured by the Azure Functions worker and forwarded to the App Insights traces table — no extra setup required.

For the exceptions table, register a tracker via set_exception_tracker(fn) (see the Quick Start). The tracker receives (exception, properties) for every Error/Warning log and is expected to forward it to App Insights. Properties set by the library:

Information severity logs never invoke the tracker.

⚠️ Do not call configure_azure_monitor() from azure-monitor-opentelemetry. It enables auto-instrumentation of requests, urllib3, httpx, and other libraries, which floods the dependencies and requests tables with Azure SDK traffic (token acquisition, Log Analytics queries, …) and emits verbose HTTP dumps as traces. Use a standalone exporter wired through set_exception_tracker() instead.

License

Private — @actvalue