Structured Logging with the Python Standard Library: Zero-Dependency JSON Output

You need machine-parseable JSON logs but cannot add a logging dependency to the service — because of an audit policy, a locked-down base image, or a library that must not force its logging stack on consumers. This page is for backend engineers and SREs who already run logging and want structured output from it, nothing more. It is part of the Python Logging Fundamentals and Structured Data guide and applies the formatter configuration material to one concrete task: subclass logging.Formatter, project the LogRecord into a deterministic dictionary, and wire it through dictConfig.

Prerequisites

Everything here ships with CPython, so there is nothing to install and nothing to pin except the interpreter itself. Python 3.10 or newer is assumed for datetime.UTC-style timezone handling and modern typing syntax; the code runs unchanged on 3.8 if you keep timezone.utc.

# pyproject.toml — the only "dependency" this technique has.
[project]
name = "payment-service"
requires-python = ">=3.10"
dependencies = []          # no logging library needed

# Optional: a service identifier the formatter can stamp on every record.
export SERVICE_NAME="payment-service"
export LOG_LEVEL="INFO"

Implementation

Step 1 — Subclass logging.Formatter and override format(). Build a dictionary from the attributes you want on the wire. Select fields explicitly rather than dumping record.__dict__, which carries internal bookkeeping (relativeCreated, msg, args, levelno) and any attribute a third-party library may have attached. An explicit projection is the only way to keep the schema stable enough for a query layer to depend on, and it keeps the serialized line small. Note that the base class's format() also fills record.message and record.asctime as a side effect when called; overriding format() outright, as here, skips that machinery and is both faster and free of the local-timezone trap that formatTime carries.

Step 2 — Translate the level into OpenTelemetry severity. Emit severity_text from record.levelname, but do not emit record.levelno as severity_number: Python numbers its levels 10/20/30/40/50 while OpenTelemetry uses 5/9/13/17/21, so shipping the Python integer under the OTel field name quietly breaks every threshold rule downstream. Translate through an explicit table, falling back to the nearest standard level below so custom levels such as a 25-point AUDIT still land on a valid severity. The canonical table, including the syslog column, lives in log levels and severity mapping. Convert the timestamp from record.created with datetime.fromtimestamp(..., tz=timezone.utc) so the field is unambiguous RFC 3339 rather than the host's local time.

Step 3 — Handle exceptions and caller-supplied fields safely. Call self.formatException(record.exc_info) and store the result as one string value. Keeping it inside json.dumps is what makes it safe: the serializer escapes the embedded newlines, so the record still occupies exactly one physical line for a line-delimited consumer. The base class caches its rendering in record.exc_text after the first call, so two handlers sharing a record do not re-render the traceback twice. Merge caller context from a dedicated extra_fields key, guarded with if k not in log_obj, so user data can extend the payload but never overwrite the reserved schema.

Step 4 — Wire the formatter through dictConfig and propagate context. Reference the formatter by its callable path, attach it to a StreamHandler, and read trace identifiers from contextvars so async tasks emit correlated records. Avoid basicConfig() in production; declarative configuration fails fast on a bad class path instead of silently logging unformatted lines.

import json
import logging
import logging.config
import asyncio
from contextvars import ContextVar
from datetime import datetime, timezone

# Async-safe context: each task and thread gets its own copy.
trace_id_var: ContextVar[str] = ContextVar("trace_id", default="")
span_id_var: ContextVar[str] = ContextVar("span_id", default="")

# Python level -> OpenTelemetry SeverityNumber, highest first.
_SEVERITY_TABLE = (
    (logging.CRITICAL, 21),
    (logging.ERROR, 17),
    (logging.WARNING, 13),
    (logging.INFO, 9),
    (logging.DEBUG, 5),
)


def severity_number(levelno: int) -> int:
    # Custom levels (e.g. AUDIT at 25) fall to the nearest standard level below.
    for threshold, severity in _SEVERITY_TABLE:
        if levelno >= threshold:
            return severity
    return 1  # TRACE / unspecified


class OTelJSONFormatter(logging.Formatter):
    def format(self, record: logging.LogRecord) -> str:
        log_obj = {
            "timestamp": datetime.fromtimestamp(record.created, tz=timezone.utc).isoformat(),
            "severity_text": record.levelname,               # OTel severity_text
            "severity_number": severity_number(record.levelno),  # OTel severity_number
            "logger": record.name,
            "message": record.getMessage(),
            "module": record.module,
            "function": record.funcName,
            "line": record.lineno,
            "trace_id": trace_id_var.get(),
            "span_id": span_id_var.get(),
        }
        # json.dumps escapes the traceback's newlines, so the record stays one line.
        if record.exc_info and record.exc_info[0]:
            log_obj["exception"] = self.formatException(record.exc_info)
        # Merge caller context without clobbering reserved keys.
        extra = getattr(record, "extra_fields", {})
        log_obj.update({k: v for k, v in extra.items() if k not in log_obj})
        return json.dumps(log_obj, default=str, sort_keys=True)


LOGGING_CONFIG = {
    "version": 1,
    "disable_existing_loggers": False,  # keep third-party loggers alive
    "formatters": {"otel_json": {"()": "__main__.OTelJSONFormatter"}},
    "handlers": {
        "console": {"class": "logging.StreamHandler", "formatter": "otel_json", "level": "INFO"}
    },
    "root": {"level": "INFO", "handlers": ["console"]},
}
logging.config.dictConfig(LOGGING_CONFIG)


async def simulate_request():
    # Populate context as if extracted from a W3C traceparent header at ingress.
    trace_id_var.set("4bf92f3577b34da6a3ce929d0e0e4736")
    span_id_var.set("00f067aa0ba902b7")
    logger = logging.getLogger("payment.service")
    logger.info("Transaction processed", extra={"extra_fields": {"amount": 49.99, "currency": "USD"}})
    try:
        raise ValueError("Invalid payment gateway response")
    except Exception:
        logger.exception("Payment routing failed")


if __name__ == "__main__":
    asyncio.run(simulate_request())

Why subclassing beats string templates

The base logging.Formatter interpolates a %-style template such as "%(levelname)s %(message)s", which can fake JSON by writing a template full of quotes and braces. That approach breaks the moment a message or a field value contains a quote, a backslash, or a newline, because string interpolation does no escaping. Subclassing and serializing with json.dumps is the only standard-library technique that produces valid JSON for arbitrary input, because the serializer escapes every value at the boundary. The default=str argument is the second half of that guarantee: it coerces a datetime, Decimal, or UUID that would otherwise raise TypeError into a string instead of crashing the formatter mid-line. That escaping guarantee is also the main thing a dependency would buy you — the rest of the standard library versus third-party trade-off is ergonomics, not capability.

Two ways to attach extra fields

There are two complementary mechanisms for getting application context into a record, and they suit different scopes. The extra= keyword on a single call carries data that changes per statement; the logging module copies each key onto the LogRecord as an attribute, so the formatter reads it back with getattr. Nest it under a single key — the example uses extra_fields — so user data can never collide with a reserved attribute like message or args, which would raise KeyError. A LoggerAdapter, by contrast, binds a fixed dict to a logger so every call from that component carries the same context without repeating it.

import logging

logger = logging.getLogger("payment.service")
# Component-scoped context: every call from this adapter carries worker_id.
worker = logging.LoggerAdapter(logger, {"extra_fields": {"worker_id": "w-3"}})
# Per-call context layered on top for this one statement.
worker.info("charge settled", extra={"extra_fields": {"amount": 12.0}})

Expected Output:

{"amount": 12.0, "message": "charge settled", "severity_number": 9, "severity_text": "INFO", "worker_id": "w-3", ...}

Injecting request context with contextvars

The formatter reads trace_id_var.get() and span_id_var.get() at serialization time, which works because a ContextVar is copied per asyncio task and per thread. Set the values once at ingress — typically in middleware that parses a W3C traceparent header — and every record emitted anywhere downstream in that task picks them up with no call-site plumbing; the full middleware recipe is in using contextvars for request tracing. The critical discipline is to reset the variable when the request ends, using the token returned by set(), so a worker that handles many requests on one thread never bleeds one request's trace ID into the next.

def begin(trace_id: str, span_id: str):
    # Return tokens so the caller can reset cleanly in a finally block.
    return trace_id_var.set(trace_id), span_id_var.set(span_id)


def end(tokens):
    t_tok, s_tok = tokens
    trace_id_var.reset(t_tok)
    span_id_var.reset(s_tok)

Expected Output:

# After end(tokens), the next record outside the request scope logs trace_id ""
{"message": "idle", "severity_number": 9, "span_id": "", "trace_id": "", ...}

If the identifiers should come from a live OpenTelemetry span rather than middleware you populate yourself, swap the two context variables for the record-factory approach in adding trace IDs to log records, and route the finished lines through a non-blocking sink as described in handler architecture.

Configuration options

Verification

Run the module with python app.py. Each line must be a single JSON object; piping through python -m json.tool confirms it parses, and wc -l confirms the traceback did not split the record across lines.

Expected Output (two newline-delimited JSON records):

{"amount": 49.99, "currency": "USD", "function": "simulate_request", "line": 53, "logger": "payment.service", "message": "Transaction processed", "module": "__main__", "severity_number": 9, "severity_text": "INFO", "span_id": "00f067aa0ba902b7", "timestamp": "2026-07-25T10:30:00+00:00", "trace_id": "4bf92f3577b34da6a3ce929d0e0e4736"}
{"exception": "Traceback (most recent call last):\n ...\nValueError: Invalid payment gateway response", "function": "simulate_request", "line": 58, "logger": "payment.service", "message": "Payment routing failed", "module": "__main__", "severity_number": 17, "severity_text": "ERROR", "span_id": "00f067aa0ba902b7", "timestamp": "2026-07-25T10:30:00+00:00", "trace_id": "4bf92f3577b34da6a3ce929d0e0e4736"}

The \n sequences inside the exception value are escaped by the serializer, which is why the second record is still one line. Assertions pin the contract in a unit test so a refactor cannot quietly change the schema:

import json, logging


def test_emits_valid_json_with_otel_severity():
    rec = logging.LogRecord("t", logging.INFO, __file__, 1, "hi", None, None)
    line = OTelJSONFormatter().format(rec)
    parsed = json.loads(line)            # raises if not valid JSON
    assert "\n" not in line              # one record, one physical line
    assert parsed["severity_number"] == 9    # OTel scale, not Python's 20
    assert parsed["severity_text"] == "INFO"


def test_extra_fields_and_context():
    # Context set in this task must surface on the record.
    trace_id_var.set("4bf92f3577b34da6a3ce929d0e0e4736")
    rec = logging.LogRecord("t", logging.INFO, __file__, 1, "ok", None, None)
    rec.extra_fields = {"amount": 5}     # simulate extra= injection
    parsed = json.loads(OTelJSONFormatter().format(rec))
    assert parsed["amount"] == 5
    assert parsed["trace_id"] == "4bf92f3577b34da6a3ce929d0e0e4736"
    # Reserved keys must win over user-supplied collisions.
    rec.extra_fields = {"message": "spoofed"}
    parsed = json.loads(OTelJSONFormatter().format(rec))
    assert parsed["message"] == "ok"     # formatter guards reserved keys

Common mistakes

• Error signature: the handler writes null for ordinary records, or raises TypeError: expected str instance, NoneType found from StreamHandler.emit. Root cause: the return json.dumps(...) statement sits inside the if record.exc_info: block, so format() falls off the end and returns None for every record without an exception. Remediation: make the serialization the final, unconditional statement of format(), and keep the test_emits_valid_json_with_otel_severity assertion above in CI so a non-exception record is always exercised.

• Error signature: --- Logging error --- appears on stderr with TypeError: Object of type datetime is not JSON serializable, and the record itself never reaches the sink. Root cause: a caller passed a non-primitive value — a datetime, Decimal, UUID, or an ORM model — through extra_fields, and json.dumps refuses it. Remediation: pass default=str as shown, and for high-value fields convert at the call site so the stored value has a deliberate shape rather than whatever str() produces.

• Error signature: the log shipper reports unexpected end of JSON input for exactly the records that carry tracebacks, and the sink shows fragments such as File "app.py", line 58. Root cause: the exception text was written outside the JSON string value — concatenated onto the formatted line, or emitted by a second handler — so its raw newlines became record separators. Remediation: keep self.formatException(...) inside the dictionary handed to json.dumps, which escapes the newlines, and verify with the "\n" not in line assertion.

• Error signature: KeyError: "Attempt to overwrite 'message' in LogRecord" is raised at the log call, not in the formatter. Root cause: extra= was given a reserved attribute name directly, as in extra={"message": "x"}; the logging module refuses to clobber its own attributes. Remediation: nest all caller data under one namespaced key such as extra_fields and merge it with the k not in log_obj guard so user values can never shadow the reserved schema.

• Error signature: identifiers look correct under light load but an idle line, or a line from the next request, carries the previous request's trace_id. Root cause: either a module-level dictionary is being used for request state — which is shared by every worker in a WSGI or ASGI process — or a ContextVar.set() had no matching reset(), leaving the value live on a reused thread. Remediation: keep request state in contextvars, capture the token from set(), and call reset(token) in a finally block at the request boundary; the thread and process boundary rules are covered in context variables and thread safety.

Related

• Formatter configuration — the parent guide on serialization, timestamps, and exception rendering.
• Configuring logging with dictConfig — the declarative wiring this formatter plugs into.
• Adding trace IDs to log records — replace the manual context variables with live OpenTelemetry span identifiers.
• Log levels and severity mapping — the canonical Python, OpenTelemetry, and syslog severity table.
• Non-blocking logging with QueueHandler — keep JSON serialization and I/O off the request thread.

Frequently Asked Questions