actvalue.azure-errors 0.0.5

Structured JSON logging and error handling for Python Azure Functions, designed for Application Insights consumption.

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

Function	Description
run_with_context(context, fn)	Run fn (sync or async) with context stored in a ContextVar
get_context()	Retrieve the current context (None outside a request scope)

Logging

Function	Description
set_function_app(name)	Set the FunctionApp field for all logs
configure_app_insights_exceptions(connection_string, *, level=WARNING)	One-call setup that wires Error/Warning logs to the App Insights exceptions table via a standalone AzureMonitorLogExporter (no auto-instrumentation). Requires the [appinsights] extra.
set_exception_tracker(fn)	Lower-level hook: register a (exception, properties) -> None callable invoked for every Error/Warning log. Pass None to disable. Use this if you need a custom exporter/client.
silence_azure_sdk_logs(level=logging.WARNING)	Drop INFO-level noise from azure.*, msal, urllib3, opentelemetry loggers so it never reaches the App Insights traces table
log_structured(log, context?)	Emit a full LogInput dict; routes to logging.info/warning/error by severity
log_info(message, code_or_details?)	Convenience — severity Information. Pass a str for Code or a dict for Details
log_warning(error_type, message, code_or_details?)	Convenience — severity Warning. Pass a str for Code or a dict for Details
log_error(error_type, message, code_or_details?)	Convenience — severity Error. Pass a str for Code or a dict for Details

Errors

Class	Status	Severity
NotFoundError	404	Warning
ForbiddenError	403	Warning
BadRequestError	400	Warning

Function	Description
on_error(context, error)	HTTP catch handler — logs structured entry and returns HttpResponse
on_queue_error(context, error)	Queue/timer catch handler — logs structured entry and re-raises for Azure runtime retry/dead-lettering

Severity Routing

Logs are routed to the appropriate Python logging level:

Severity	Logging level
Error	logging.error()
Warning	logging.warning()
Information	logging.info()

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:

Property	Value
FunctionApp	from set_function_app()
FunctionName	from context
InvocationId	from context
Severity	Error or Warning
ErrorType	from the log entry
Code	included when provided
Details	JSON-stringified, included when provided

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