Python stdlib logging

A logging.Handler that ships stdlib log records to axonpush as OpenTelemetry-shaped app.log events.

pip install axonpush

No extra required, the handler is stdlib-only. It works with anything built on Python’s logging: FastAPI, Flask, Django, Celery, aiohttp, Starlette.

Attach the handler

import logging
from axonpush import AxonPush
from axonpush.integrations.logging_handler import AxonPushLoggingHandler

client = AxonPush()

handler = AxonPushLoggingHandler(
    client=client,
    channel_id=channel_id,
    service_name="my-api",
    environment="production",
)

root = logging.getLogger()
root.addHandler(handler)
root.setLevel(logging.INFO)

Every argument is keyword-only, including channel_id.

Then log normally. extra={...} becomes OTel attributes:

log = logging.getLogger("my_app.orders")

log.info("order created", extra={"order_id": 1234, "total": 49.99})
log.warning("stock low for sku=%s", "A-42", extra={"remaining": 3})

try:
    charge(card)
except RuntimeError:
    log.exception("failed to charge card", extra={"order_id": 1234})

Constructor

AxonPushLoggingHandler(
    *,
    channel_id,                  # str UUID - required
    client=None,                 # AxonPush or AsyncAxonPush
    api_key=None,                # or let it read AXONPUSH_API_KEY
    tenant_id=None,              # or AXONPUSH_TENANT_ID
    base_url=None,               # or AXONPUSH_BASE_URL
    source="app",                # "app" -> app.log, "agent" -> agent.log
    service_name=None,
    service_version=None,
    environment=None,
    agent_id=None,
    level=logging.NOTSET,
    exclude_loggers=None,        # extra logger-name prefixes to drop
    mode=None,                   # "background" (default) | "sync"
    queue_size=1000,
    shutdown_timeout=2.0,
)

Pass either client= or the credential keywords, passing both raises ValueError. With neither, the handler builds its own client from AXONPUSH_API_KEY / AXONPUSH_TENANT_ID / AXONPUSH_BASE_URL, which is what makes the Django dictConfig path below possible.

mode="background" is the default: emit() pushes onto a bounded queue and returns, and a daemon thread does the publishing. mode="sync" publishes on the calling thread, useful in tests and one-shot scripts, wrong in a request handler.

mode="background" only starts a worker for a sync AxonPush client. Give the handler an AsyncAxonPush and every record is published inline instead, pass a plain AxonPush, which is safe here precisely because emit() never touches the network on your thread.

The recursion filter

Publishing a record makes an HTTP request, and httpx logs that request, which would be captured and published, and so on. The handler installs a filter that drops records from httpx, httpcore, axonpush and axonpush.publisher, plus a context-var check that discards anything emitted while the publisher is mid-flight.

The filter is always on. exclude_loggers=[...] adds your own prefixes to it:

AxonPushLoggingHandler(
    client=client,
    channel_id=channel_id,
    exclude_loggers=["werkzeug", "django.db.backends"],
)

What each record becomes

Field	Value
`identifier`	The logger name, e.g. `my_app.orders`
`event_type`	`app.log`, or `agent.log` when `source="agent"`
`payload.body`	The formatted message
`payload.severityNumber` / `payload.severityText`	OTel severity from the Python level: DEBUG 5, INFO 9, WARN 13, ERROR 17, FATAL 21
`payload.timeUnixNano`	`record.created`, in nanoseconds
`payload.attributes`	`code.filepath`, `code.function`, `code.lineno`, `code.namespace`, `logger.name`, `thread.name`, `process.pid`, plus every `extra={...}` key
`payload.resource`	`service.name`, `service.version`, `deployment.environment` when configured

A record carrying exc_info also gets exception.type and exception.message.

The trace id comes from the active trace context, so log lines emitted inside a traced request join that trace.

Flushing

handler.flush(timeout=1.0)   # block until the queue drains, or give up
handler.close()              # drain, stop the worker, release the handler

logging.shutdown() calls close() at interpreter exit, so a long-running process needs neither explicitly.

Framework notes

FastAPI, Starlette, uvicorn

Build the handler in main.py before FastAPI() so startup logs go through it.

import logging, os
from fastapi import FastAPI
from axonpush import AxonPush
from axonpush.integrations.logging_handler import AxonPushLoggingHandler

client = AxonPush()

handler = AxonPushLoggingHandler(
    client=client,
    channel_id=os.environ["AXONPUSH_CHANNEL_ID"],
    service_name="my-api",
)
handler.setLevel(logging.INFO)

logging.getLogger().addHandler(handler)
logging.getLogger("uvicorn.error").addHandler(handler)
logging.getLogger().setLevel(logging.INFO)

app = FastAPI()

The second addHandler is not redundant. Uvicorn’s LOGGING_CONFIG sets propagate=False on uvicorn and uvicorn.access, so records on uvicorn.error fire uvicorn’s own handler and then stop, they never reach root. A root-only attach silently misses most of what you wrote in main.py. Attach to uvicorn.error, not to uvicorn, or root and uvicorn both fire and you get every record twice.

Add logging.getLogger("uvicorn.access").addHandler(handler) if you want one event per HTTP request.

Django

dictConfig only accepts primitive values, so you cannot hand it a pre-built client. Give it credentials instead, or set the environment variables and omit them entirely.

# settings.py
import os

LOGGING = {
    "version": 1,
    "disable_existing_loggers": False,
    "handlers": {
        "console": {"class": "logging.StreamHandler"},
        "axonpush": {
            "class": "axonpush.integrations.logging_handler.AxonPushLoggingHandler",
            "channel_id": os.environ["AXONPUSH_CHANNEL_ID"],
            "service_name": "my-django-app",
            "environment": "production",
            "exclude_loggers": ["django.db.backends"],
        },
    },
    "root": {"handlers": ["console", "axonpush"], "level": "INFO"},
}

django.db.backends logs every SQL query at DEBUG. Exclude it, or pin its level, unless you want the channel to be mostly SQL.

Flask

Same root-logger pattern, or attach to app.logger for Flask-scoped records only:

app.logger.addHandler(handler)

Werkzeug’s dev-server access logs are not excluded by default, they are no recursion risk. Add exclude_loggers=["werkzeug"] if you do not want them.

Gunicorn preload and Celery prefork

The background publisher registers an os.register_at_fork hook, so a forked child gets a fresh queue and a fresh worker thread rather than inheriting the parent’s. Constructing the handler once at module level is fine under gunicorn --preload and celery --pool=prefork.

Lambda, Cloud Functions, Azure Functions

Serverless containers freeze between invocations, so a background thread may never get to drain. Flush at the end of each invocation:

import logging, os
from axonpush import AxonPush
from axonpush.integrations.logging_handler import (
    AxonPushLoggingHandler,
    flush_after_invocation,
)

client = AxonPush()
handler = AxonPushLoggingHandler(
    client=client,
    channel_id=os.environ["AXONPUSH_CHANNEL_ID"],
    service_name="my-lambda",
)
logging.getLogger().addHandler(handler)
logging.getLogger().setLevel(logging.INFO)

@flush_after_invocation(handler)
def lambda_handler(event, context):
    logging.info("processing event", extra={"event_id": event["id"]})
    return {"statusCode": 200}

flush_after_invocation flushes in a finally: block, once per invocation, not once per log line. It accepts several handlers at once, and the same decorator works for the Loguru sink, the structlog processor and the OTel exporter:

@flush_after_invocation(log_handler, otel_exporter, loguru_sink, timeout=5.0)
def lambda_handler(event, context):
    ...

The handler detects Lambda via AWS_LAMBDA_FUNCTION_NAME, Cloud Functions via FUNCTION_TARGET and Azure Functions via AZURE_FUNCTIONS_ENVIRONMENT, and logs a one-time reminder at construction.

When the queue overflows

The queue holds queue_size records (1000 by default) and drops the oldest when full, counting the drops and warning on the axonpush.publisher logger at most once every ten seconds. Raise queue_size if you log in bursts.

On this page