HTTPX Logging Handler Example

Use this reference when an application should emit JSON logs to an HTTP collector while keeping startup logging configuration declarative.

This page follows the top-level skill pattern:

define one LOGGING dictionary
apply it once with logging.config.dictConfig(LOGGING)
keep modules focused on logger calls

Source docs to keep nearby:

Minimal Topology

application code -> named logger -> HttpxJsonLogHandler -> HTTP collector

Reusable Handler Type

Keep transport behavior in one handler class and wire it declaratively through dictConfig.

import logging

import httpx


class HttpxJsonLogHandler(logging.Handler):
    def __init__(self, collector_url: str, timeout_seconds: float = 2.0, token: str | None = None) -> None:
        super().__init__()
        headers = {"content-type": "application/json"}
        if token is not None:
            headers["authorization"] = f"Bearer {token}"
        timeout = httpx.Timeout(timeout_seconds)
        self.client = httpx.Client(base_url=collector_url, headers=headers, timeout=timeout)

    def emit(self, record: logging.LogRecord) -> None:
        payload = {
            "name": record.name,
            "levelname": record.levelname,
            "levelno": record.levelno,
            "pathname": record.pathname,
            "lineno": record.lineno,
            "funcName": record.funcName,
            "created": record.created,
            "message": record.getMessage(),
        }
        try:
            response = self.client.post("/logs", json=payload)
            response.raise_for_status()
        except httpx.HTTPError:
            self.handleError(record)

    def close(self) -> None:
        self.client.close()
        super().close()

Application Logging Configuration (Declarative)

import logging.config

LOGGING = {
    "version": 1,
    "disable_existing_loggers": False,
    "handlers": {
        "httpx": {
            "class": "httpx_json_handler.HttpxJsonLogHandler",
            "collector_url": "http://127.0.0.1:9021",
            "timeout_seconds": 2.0,
            "token": None,
        },
        "console": {
            "class": "logging.StreamHandler",
            "level": "INFO",
            "stream": "ext://sys.stdout",
        },
    },
    "root": {
        "level": "INFO",
        "handlers": ["httpx", "console"],
    },
}


def configure_logging() -> None:
    logging.config.dictConfig(LOGGING)

import logging

logger = logging.getLogger(__name__)


def sync_customer(customer_id: str) -> None:
    logger.info("Syncing customer %s", customer_id)

from feature import sync_customer
from logging_config import configure_logging

configure_logging()
sync_customer("C-101")

Collector-Side Configuration (Declarative)

Whether you use an internal HTTP endpoint or a managed collector, keep receiver-side formatting and routing declared on the receiver side, not in application modules.

Why This Pattern

Logging wiring is declared once and applied once.
Runtime behavior changes by editing config fields, not scattered root mutations.
Feature modules stay independent from transport details.
HTTP connection details remain encapsulated in one handler type.

Review Checklist

Is there one LOGGING dict for the application process?
Is dictConfig called once at startup?
Are module loggers created via logging.getLogger(__name__)?
Are HTTP endpoint, timeout, and auth token inputs declared in handler config?
Are final routing/retention decisions handled by the collector side?

3.9 KiB Raw Blame History