Create custom log handlers¶
The question. You need log records to go somewhere the built-in handlers don't reach — an in-memory list for assertions in a test, a CSV file for later analysis, a Slack webhook, a metrics system. The stock StreamHandler, FileHandler, and RotatingFileHandler don't cover it.
The answer: subclass logging.Handler and override emit(record). The base class takes care of levels, formatters, and thread-safe dispatch; you just decide what to do with the already-formatted message.
# The minimal shape: one class, one overridden emit().
# This ListHandler captures messages in-process — useful for tests and for
# programmatic inspection of what got logged.
import logging
class ListHandler(logging.Handler):
'''Captures formatted log messages in a list (great for tests).'''
def __init__(self, level: int = logging.NOTSET) -> None:
super().__init__(level)
self.messages: list[str] = []
def emit(self, record: logging.LogRecord) -> None:
try:
self.messages.append(self.format(record))
except Exception:
self.handleError(record) # never let handler errors escape
# Usage — attach to any logger like a built-in handler.
logger = logging.getLogger('demo')
logger.setLevel(logging.DEBUG)
handler = ListHandler()
handler.setFormatter(logging.Formatter('%(levelname)s: %(message)s'))
logger.addHandler(handler)
logger.info('first message')
logger.warning('second message')
logger.error('third message')
for msg in handler.messages:
print(' ', msg)
logger.removeHandler(handler)
Variant: a CSV handler¶
Writes each record as a row: timestamp, level, logger name, message. Useful for post-hoc analysis in pandas or Excel without parsing a free-form log format.
import csv
import logging
from datetime import datetime, timezone
from pathlib import Path
class CSVHandler(logging.Handler):
'''Writes records as CSV rows — one row per emit.'''
def __init__(self, path: str | Path, level: int = logging.NOTSET) -> None:
super().__init__(level)
self.path = Path(path)
with self.path.open('w', newline='', encoding='utf-8') as f:
csv.writer(f).writerow(['timestamp', 'level', 'logger', 'message'])
def emit(self, record: logging.LogRecord) -> None:
try:
ts = datetime.fromtimestamp(record.created, tz=timezone.utc).isoformat()
with self.path.open('a', newline='', encoding='utf-8') as f:
csv.writer(f).writerow([ts, record.levelname, record.name, record.getMessage()])
except Exception:
self.handleError(record)
import tempfile
csv_path = Path(tempfile.mkdtemp()) / 'audit.csv'
logger = logging.getLogger('csv-demo')
logger.setLevel(logging.DEBUG)
logger.addHandler(CSVHandler(csv_path))
logger.info('service started')
logger.warning('cache miss for key %r', 'user:42')
print(csv_path.read_text())
logger.handlers.clear()
Variant: filters for content-based routing¶
Attach a logging.Filter to a handler (or a logger) to accept/reject records based on anything you can compute from the record. Here, only records whose message contains 'security' get through.
import logging
class KeywordFilter(logging.Filter):
def __init__(self, keyword: str) -> None:
super().__init__()
self.keyword = keyword
def filter(self, record: logging.LogRecord) -> bool:
return self.keyword in record.getMessage()
class ListHandler(logging.Handler):
def __init__(self, level: int = logging.NOTSET) -> None:
super().__init__(level)
self.messages: list[str] = []
def emit(self, record: logging.LogRecord) -> None:
self.messages.append(self.format(record))
logger = logging.getLogger('filter-demo')
logger.setLevel(logging.DEBUG)
handler = ListHandler()
handler.setFormatter(logging.Formatter('%(levelname)s: %(message)s'))
handler.addFilter(KeywordFilter('security'))
logger.addHandler(handler)
logger.info('user logged in') # filtered out
logger.warning('security: failed login attempt') # kept
logger.info('data processed') # filtered out
logger.error('security: unauthorised access attempt') # kept
for m in handler.messages:
print(' ', m)
logger.handlers.clear()
Why this works¶
Every handler inherits from logging.Handler, and the framework's whole job is to turn a LogRecord into a call to emit(record). Level filtering, formatter application, thread safety, and locking all happen in the base class. You only need to decide what to do once the record arrives — so emit is usually three or four lines.
self.format(record) does the formatter dance for you: applies the format string you set with setFormatter, renders %(asctime)s/%(levelname)s/etc., and returns a string. For destinations that want structured data rather than a rendered string, reach directly into the record attributes — record.created, record.levelname, record.getMessage(), record.exc_info — and skip format() entirely.
Wrapping the body in try / self.handleError(record) is the standard defensive shape. If your CSV file is suddenly unwritable or the network handler times out, you want the failure logged by the framework rather than raised into the caller's code path.
Trade-offs¶
Reach for a custom handler only when the built-ins really don't fit. StreamHandler, FileHandler, RotatingFileHandler, TimedRotatingFileHandler, SMTPHandler, and SysLogHandler between them cover most production needs. MemoryHandler even covers the in-process buffer case if you don't need an assertable list.
For filtering by content (not level), add a logging.Filter rather than baking the logic into emit. Filters compose — one on the logger, one on each handler — and keep emit focused on the destination.
When the handler does I/O that could block (HTTP, databases, slow disks), consider using QueueHandler on the app side with a background thread running the real handler. That way your request handler isn't waiting on the log system.
Related reading¶
- Configure logging for a project — how to wire your custom handler into
dictConfig. - Avoid common logging mistakes — handler-level vs. logger-level, duplicate handlers.
- Logging module quick reference — the full list of built-in handlers.