Writing a Custom Logging Handler in Python

Eventually a sink exists that no stdlib handler covers — an internal audit bus, a vendor SDK, a socket with its own framing. This page covers the four things a custom logging.Handler must get right: what belongs in emit(), why it must never raise, where the lock is, and what close() owes the interpreter. It builds on the handler architecture guide and is part of the Python logging fundamentals and structured data section.

The base class does more than it looks like it does. Most handler bugs come from re-implementing something Handler already handled, or from doing something inside emit() that was never meant to be there.

Prerequisites

Standard library only for the handler itself; the example ships records to an HTTP endpoint, so pin a client if you follow it literally.

pip install "httpx>=0.27.0,<1.0.0"

export AUDIT_SINK_URL="https://audit.internal/v1/events"
export AUDIT_BUFFER_MAX=2000

Implementation

Step 1 — Override emit() and nothing else. Format through self.format(record) so the handler's formatter, and any formatter someone sets later through dictConfig, is respected. Wrap the body so nothing escapes.

import logging

class AuditHandler(logging.Handler):
    def __init__(self, url: str, level: int = logging.NOTSET):
        super().__init__(level=level)
        self.url = url

    def emit(self, record: logging.LogRecord) -> None:
        try:
            payload = self.format(record)          # honours whatever formatter is set
            self._enqueue(payload)                 # must be cheap — see step 2
        except Exception:
            self.handleError(record)               # never propagate into caller code

handleError() is the contract. Logging is infrastructure; an audit endpoint returning 503 must not turn into a 500 for the user whose request happened to log. The base implementation prints to stderr when logging.raiseExceptions is true, which is right for development and wrong for production — set it to False and, if you need visibility, override handleError to increment a counter instead.

Step 2 — Get the I/O out of the lock. Handler.handle() holds self.lock for the whole of emit(). An HTTP call there does not just slow that one log statement; it stops every thread in the process that logs to this handler for the duration. Buffer in emit(), send from a worker.

import queue
import threading
import httpx

class AuditHandler(logging.Handler):
    def __init__(self, url: str, max_buffer: int = 2000, batch: int = 100):
        super().__init__()
        self.url = url
        self.batch = batch
        self._buffer: queue.Queue[str] = queue.Queue(maxsize=max_buffer)
        self._stopping = threading.Event()
        self._worker = threading.Thread(target=self._run, name="audit-sink", daemon=True)
        self._worker.start()

    def emit(self, record: logging.LogRecord) -> None:
        try:
            self._buffer.put_nowait(self.format(record))   # never blocks the caller
        except queue.Full:
            self.handleError(record)                       # explicit drop, counted
        except Exception:
            self.handleError(record)

    def _run(self) -> None:
        with httpx.Client(timeout=5.0) as client:
            while not (self._stopping.is_set() and self._buffer.empty()):
                batch = []
                try:
                    batch.append(self._buffer.get(timeout=0.5))
                except queue.Empty:
                    continue
                while len(batch) < self.batch:
                    try:
                        batch.append(self._buffer.get_nowait())
                    except queue.Empty:
                        break
                try:
                    client.post(self.url, json={"events": batch})
                except Exception:
                    pass                                   # a sink outage must not kill the worker

The daemon=True thread plus an explicit drain in close() is deliberate: a non-daemon worker that never exits hangs the interpreter, and a daemon worker with no drain loses the buffer. You need both halves.

Step 3 — Make close() flush and deregister. logging.shutdown() walks every handler at exit and calls flush() then close(). Your override drains, stops the worker, and then calls super().close(), which removes the handler from the internal list so shutdown does not touch it twice.

    def flush(self) -> None:
        deadline = time.monotonic() + 2.0
        while not self._buffer.empty() and time.monotonic() < deadline:
            time.sleep(0.02)

    def close(self) -> None:
        try:
            self._stopping.set()
            self._worker.join(timeout=3.0)         # bounded — never block shutdown forever
        finally:
            super().close()                        # deregisters from logging's handler list

Every wait here is bounded. An unbounded join in close() is how a process that is otherwise ready to exit sits for minutes waiting on a sink that is already gone.

Step 4 — Wire it declaratively. A handler that can only be constructed in code cannot be reconfigured per environment. Reference it by dotted path in dictConfig, exactly as described in configuring logging with dictConfig.

CONFIG = {
    "version": 1,
    "disable_existing_loggers": False,
    "formatters": {"json": {"()": "pythonjsonlogger.jsonlogger.JsonFormatter"}},
    "handlers": {
        "audit": {
            "()": "observability.audit.AuditHandler",
            "url": "https://audit.internal/v1/events",
            "max_buffer": 2000,
            "level": "INFO",
            "formatter": "json",
        },
    },
    "loggers": {"audit": {"level": "INFO", "handlers": ["audit"], "propagate": False}},
}

Configuration options

Verification

Test the two properties that distinguish a safe handler from a dangerous one: it never raises, and it never blocks.

import logging, time

def test_emit_never_raises():
    h = AuditHandler(url="http://127.0.0.1:1")     # nothing listening
    h.setFormatter(logging.Formatter("%(message)s"))
    record = logging.LogRecord("t", logging.INFO, "", 0, "hello", (), None)
    h.emit(record)                                  # must not raise
    h.close()

def test_emit_is_fast_even_when_the_sink_is_down():
    h = AuditHandler(url="http://127.0.0.1:1")
    h.setFormatter(logging.Formatter("%(message)s"))
    record = logging.LogRecord("t", logging.INFO, "", 0, "hello", (), None)
    start = time.perf_counter()
    for _ in range(1000):
        h.emit(record)
    assert time.perf_counter() - start < 0.1        # enqueue only, no round trips
    h.close()

Expected Output:

test_handler.py::test_emit_never_raises PASSED
test_handler.py::test_emit_is_fast_even_when_the_sink_is_down PASSED

Common mistakes

The handler blocks every thread in the process

Error signature: request latency across unrelated endpoints tracks the log sink's health. Root cause: emit() performs network I/O while holding the handler lock. Remediation: buffer in emit(), send from a worker — or keep the handler simple and put a QueueHandler in front of it, as in non-blocking logging with QueueHandler.

An exception from the sink reaches application code

Error signature: a request fails with a ConnectionError whose traceback runs through logger.info. Root cause: emit() let an exception escape instead of routing it to handleError. Remediation: wrap the whole body and call self.handleError(record). Keep logging.raiseExceptions = False in production so the fallback is silent rather than a stderr flood.

The process hangs at exit

Error signature: the container takes its full termination grace period to stop, every time. Root cause: a non-daemon worker thread, or an unbounded join() in close(), waiting on a sink that is unreachable. Remediation: daemon worker, bounded drain, bounded join, and super().close() at the end.

When not to write one

A custom handler is the right answer less often than it looks, and the alternatives are cheaper to operate. Three of them cover most cases that reach for a subclass first.

Write to stdout and let something else ship it. In a containerised deployment, a StreamHandler to stdout plus a collector agent is fewer moving parts than any in-process shipper: no buffer to size, no worker thread to shut down, no retry policy to get wrong, and a sink outage becomes the platform's problem rather than your process's. The custom handler earns its place when the destination genuinely cannot be reached that way — an internal audit bus with its own protocol, a vendor SDK with no file or stdout mode.

Use a QueueHandler in front of a simple handler. If the only reason for the subclass is to get I/O off the calling thread, the standard library already has that: a QueueHandler on the logger and the concrete handler owned by a QueueListener. That gives the background-thread behaviour without writing or maintaining the worker, and it composes with every other stdlib handler.

Use a filter or a formatter instead. A surprising share of custom handlers exist to change what a record looks like or which records are emitted, both of which belong in a formatter or a filter. Those are smaller, individually testable, and composable across sinks — a formatter written once serves every handler, while behaviour embedded in a handler subclass is available only to that one.

If you do write one, keep the surface small

The handlers that age well share a shape: they know how to send one batch to one destination, and they know nothing about formatting, filtering, levels, or what should be logged. Everything else is composed around them.

Two habits reinforce that. First, take the destination as a constructor argument rather than reading configuration inside the class, so the handler is testable with a fake and configurable with dictConfig. Second, expose the operational state — records buffered, records dropped, last error — as attributes a metrics callback can read, rather than logging about itself. A handler that logs its own failures through the logging system it is part of will, eventually, produce a loop.

class AuditHandler(logging.Handler):
    def __init__(self, url: str, max_buffer: int = 2000):
        super().__init__()
        self.dropped = 0                 # read by a metrics callback, not logged
        self.last_error: str | None = None
        ...

    def handleError(self, record: logging.LogRecord) -> None:
        self.dropped += 1                # count, do not narrate
        self.last_error = repr(sys.exc_info()[1])

Testing the failure paths

The happy path of a custom handler is easy and rarely the problem. Three tests cover the cases that actually break production, and all three are fast because none of them needs a real sink: emit with an unreachable destination must not raise and must return promptly; close with the worker blocked must return within its timeout; and a full buffer must drop rather than grow. Writing those three first tends to produce a better handler than writing them afterwards, because each one constrains the design.

Related

• Handler architecture for Python logging — the parent guide: one handler per sink, filters, and the queue boundary.
• Buffering log records with MemoryHandler — the stdlib buffering handler and when it is enough.
• Non-blocking logging with QueueHandler — the alternative to building your own worker.
• Configuring logging with dictConfig — registering a custom handler declaratively.
• Implementing custom sinks in Loguru — the same problem in Loguru's sink model.

Frequently Asked Questions