Structured logging for Python that emphasizes simplicity, power, and performance
Advanced exception processing including structured traceback extraction, rich formatting, and JSON-serializable exception representations. These tools provide comprehensive support for capturing, formatting, and analyzing exceptions in structured logging scenarios.
Data classes for representing structured exception and traceback information.
class Frame:
"""
Dataclass representing a single stack frame.
Contains information about a single frame in a stack trace,
including filename, line number, function name, and local variables.
"""
filename: str
"""Path to the source file."""
lineno: int
"""Line number in the source file."""
name: str
"""Function or method name."""
locals: dict[str, str] | None = None
"""Local variables in the frame (if captured)."""
class SyntaxError_:
"""
Dataclass representing SyntaxError details.
Specific information for syntax errors, including the problematic
line and character offset.
"""
offset: int
"""Character offset where syntax error occurred."""
filename: str
"""Filename where syntax error occurred."""
line: str
"""The problematic line of code."""
lineno: int
"""Line number where syntax error occurred."""
msg: str
"""Syntax error message."""
class Stack:
"""
Dataclass representing an exception and its stack frames.
Contains complete information about a single exception including
its type, value, notes, and the stack frames leading to it.
"""
exc_type: str
"""Exception type name."""
exc_value: str
"""String representation of exception value."""
exc_notes: list[str] = field(default_factory=list)
"""Exception notes (Python 3.11+)."""
syntax_error: SyntaxError_ | None = None
"""SyntaxError details if applicable."""
is_cause: bool = False
"""True if this exception was the cause of another."""
frames: list[Frame] = field(default_factory=list)
"""Stack frames for this exception."""
is_group: bool = False
"""True if this is part of an exception group."""
exceptions: list[Trace] = field(default_factory=list)
"""Nested exceptions for exception groups."""
class Trace:
"""
Container for complete stack trace information.
Top-level container that holds all the stacks for a complete
exception trace, including chained exceptions.
"""
stacks: list[Stack]
"""List of Stack objects representing the complete trace."""Functions and classes for extracting and processing exception information.
def extract(
exc_type,
exc_value,
traceback,
*,
show_locals=False,
locals_max_length=10,
locals_max_string=80,
locals_hide_dunder=True,
locals_hide_sunder=False,
use_rich=True
) -> Trace:
"""
Extract structured exception information from exc_info tuple.
Args:
exc_type: Exception type
exc_value: Exception instance
traceback: Traceback object
show_locals (bool): Include local variables in frames
locals_max_length (int): Maximum number of locals to show per frame
locals_max_string (int): Maximum length of local variable strings
locals_hide_dunder (bool): Hide dunder variables (__var__)
locals_hide_sunder (bool): Hide sunder variables (_var)
use_rich (bool): Use rich library for enhanced extraction if available
Returns:
Trace: Structured representation of the exception trace
"""
class ExceptionDictTransformer:
"""
Transform exceptions into JSON-serializable dictionaries.
Converts exception information into structured dictionaries
that can be easily serialized and analyzed.
"""
def __init__(
self,
show_locals=True,
locals_max_length=10,
locals_max_string=80,
locals_hide_dunder=True,
locals_hide_sunder=False,
suppress=(),
max_frames=50,
use_rich=True
):
"""
Initialize ExceptionDictTransformer.
Args:
show_locals (bool): Include local variables in output
locals_max_length (int): Maximum number of locals per frame
locals_max_string (int): Maximum length of local variable strings
locals_hide_dunder (bool): Hide dunder variables
locals_hide_sunder (bool): Hide sunder variables
suppress (tuple): Module names to suppress in tracebacks
max_frames (int): Maximum number of frames to include
use_rich (bool): Use rich library if available
"""
def __call__(self, exc_info) -> list[dict[str, Any]]:
"""
Transform exception info to structured dictionaries.
Args:
exc_info: Exception info tuple (type, value, traceback)
Returns:
list: List of dictionaries representing the exception trace
"""Helper functions for safe string conversion and representation.
def safe_str(obj) -> str:
"""
Safely convert object to string representation.
Handles cases where str() might raise an exception by
providing fallback representations.
Args:
obj: Object to convert to string
Returns:
str: String representation of the object
"""
def to_repr(obj, max_length=None, max_string=None, use_rich=True) -> str:
"""
Safe repr conversion with length limits and rich formatting.
Args:
obj: Object to represent
max_length (int, optional): Maximum length of result string
max_string (int, optional): Maximum length for string values
use_rich (bool): Use rich library for enhanced formatting
Returns:
str: String representation with length limits applied
"""import structlog
from structlog import tracebacks
def problematic_function():
local_var = "important data"
user_data = {"id": 123, "name": "Alice"}
raise ValueError("Something went wrong with the data")
try:
problematic_function()
except Exception:
import sys
# Extract structured exception information
trace = tracebacks.extract(*sys.exc_info(), show_locals=True)
# Examine the trace structure
for stack in trace.stacks:
print(f"Exception: {stack.exc_type}")
print(f"Message: {stack.exc_value}")
for frame in stack.frames:
print(f" File: {frame.filename}:{frame.lineno}")
print(f" Function: {frame.name}")
if frame.locals:
print(f" Locals: {frame.locals}")import structlog
from structlog import tracebacks, processors
import json
# Configure processor to transform exceptions to dictionaries
transformer = tracebacks.ExceptionDictTransformer(
show_locals=True,
locals_max_length=5,
locals_max_string=100
)
def exception_to_dict_processor(logger, method_name, event_dict):
"""Custom processor to handle exceptions."""
if "exc_info" in event_dict:
exc_info = event_dict.pop("exc_info")
if exc_info and exc_info != (None, None, None):
event_dict["exception_trace"] = transformer(exc_info)
return event_dict
structlog.configure(
processors=[
processors.TimeStamper(),
exception_to_dict_processor,
processors.JSONRenderer()
],
wrapper_class=structlog.BoundLogger,
)
logger = structlog.get_logger()
def nested_function():
data = {"items": [1, 2, 3], "config": {"debug": True}}
raise KeyError("Missing required key 'user_id'")
def calling_function():
context = "user_processing"
nested_function()
try:
calling_function()
except Exception:
logger.exception("Processing failed", component="user_service")
# Output includes structured exception trace as JSONimport structlog
from structlog import tracebacks, processors
def custom_exception_processor(logger, method_name, event_dict):
"""Custom processor for detailed exception formatting."""
if "exc_info" in event_dict:
exc_info = event_dict.pop("exc_info")
if exc_info and exc_info != (None, None, None):
# Extract detailed trace information
trace = tracebacks.extract(
*exc_info,
show_locals=True,
locals_max_length=3,
locals_hide_dunder=True
)
# Create custom exception summary
exception_summary = {
"error_type": trace.stacks[-1].exc_type,
"error_message": trace.stacks[-1].exc_value,
"stack_depth": sum(len(stack.frames) for stack in trace.stacks),
"top_frame": {
"file": trace.stacks[-1].frames[-1].filename,
"line": trace.stacks[-1].frames[-1].lineno,
"function": trace.stacks[-1].frames[-1].name
}
}
event_dict["exception_summary"] = exception_summary
return event_dict
structlog.configure(
processors=[
processors.TimeStamper(),
custom_exception_processor,
processors.JSONRenderer()
],
wrapper_class=structlog.BoundLogger,
)
logger = structlog.get_logger()
def process_data(data):
if not data:
raise ValueError("Data cannot be empty")
return len(data)
try:
result = process_data(None)
except Exception:
logger.exception("Data processing failed", operation="length_calculation")import structlog
from structlog import tracebacks
def analyze_exception_chain():
"""Demonstrate exception chaining analysis."""
def database_operation():
raise ConnectionError("Database connection failed")
def service_operation():
try:
database_operation()
except ConnectionError as e:
raise RuntimeError("Service operation failed") from e
def api_handler():
try:
service_operation()
except RuntimeError as e:
raise ValueError("API request failed") from e
try:
api_handler()
except Exception:
import sys
# Extract the complete exception chain
trace = tracebacks.extract(*sys.exc_info())
print(f"Exception chain has {len(trace.stacks)} levels:")
for i, stack in enumerate(trace.stacks):
print(f" Level {i + 1}: {stack.exc_type} - {stack.exc_value}")
print(f" Is cause: {stack.is_cause}")
print(f" Frames: {len(stack.frames)}")
analyze_exception_chain()import structlog
from structlog import tracebacks, processors, dev
# Configure with rich exception formatting
def rich_exception_processor(logger, method_name, event_dict):
"""Processor that uses rich for exception formatting."""
if "exc_info" in event_dict:
exc_info = event_dict.pop("exc_info")
if exc_info and exc_info != (None, None, None):
# Use ExceptionDictTransformer with rich enabled
transformer = tracebacks.ExceptionDictTransformer(
show_locals=True,
use_rich=True,
locals_max_length=5
)
event_dict["rich_exception"] = transformer(exc_info)
return event_dict
structlog.configure(
processors=[
processors.TimeStamper(),
rich_exception_processor,
processors.JSONRenderer(indent=2)
],
wrapper_class=structlog.BoundLogger,
)
logger = structlog.get_logger()
def complex_operation():
data = {"users": [{"id": 1, "name": "Alice"}], "config": {"retry": 3}}
items = [1, 2, 3, 4, 5]
# Simulate complex operation with multiple locals
for i, item in enumerate(items):
if item == 4:
raise IndexError(f"Invalid item at index {i}: {item}")
try:
complex_operation()
except Exception:
logger.exception("Complex operation failed", operation_id="op_123")from structlog import tracebacks
class ProblematicObject:
"""Object that raises exception in __str__."""
def __str__(self):
raise RuntimeError("Cannot convert to string")
def __repr__(self):
return "ProblematicObject(broken)"
# Test safe string conversion
obj = ProblematicObject()
# This would raise an exception:
# str(obj)
# But this handles it safely:
safe_string = tracebacks.safe_str(obj)
print(f"Safe string: {safe_string}")
# Test safe repr with limits
long_dict = {f"key_{i}": f"value_{i}" * 100 for i in range(10)}
limited_repr = tracebacks.to_repr(long_dict, max_length=200)
print(f"Limited repr: {limited_repr}")import structlog
from structlog import tracebacks
# Configure transformer to suppress certain modules
transformer = tracebacks.ExceptionDictTransformer(
show_locals=False,
suppress=("logging", "structlog._config", "structlog.processors"),
max_frames=10
)
def application_code():
"""User application code."""
library_code()
def library_code():
"""Simulated library code that should be suppressed."""
raise ValueError("Library error")
try:
application_code()
except Exception:
import sys
# Transform with suppression
result = transformer(sys.exc_info())
# Examine which frames were included/suppressed
for trace_dict in result:
print(f"Exception: {trace_dict.get('exc_type')}")
print(f"Frames shown: {len(trace_dict.get('frames', []))}")
for frame in trace_dict.get('frames', []):
print(f" {frame['filename']}:{frame['lineno']} in {frame['name']}")import structlog
from structlog import tracebacks, processors
class StructuredExceptionProcessor:
"""Processor that creates structured exception data."""
def __init__(self):
self.transformer = tracebacks.ExceptionDictTransformer(
show_locals=True,
locals_max_length=3,
max_frames=20
)
def __call__(self, logger, method_name, event_dict):
if "exc_info" in event_dict:
exc_info = event_dict.pop("exc_info")
if exc_info and exc_info != (None, None, None):
# Create structured exception data
trace_data = self.transformer(exc_info)
# Extract key information
if trace_data:
last_exception = trace_data[-1]
event_dict.update({
"error": {
"type": last_exception["exc_type"],
"message": last_exception["exc_value"],
"traceback": trace_data
}
})
return event_dict
structlog.configure(
processors=[
processors.TimeStamper(),
StructuredExceptionProcessor(),
processors.JSONRenderer()
],
wrapper_class=structlog.BoundLogger,
)
logger = structlog.get_logger()
def business_logic(user_data):
if not user_data.get("email"):
raise ValueError("Email is required for user registration")
try:
business_logic({"name": "Alice"})
except Exception:
logger.exception(
"User registration failed",
user_name="Alice",
registration_step="validation"
)Install with Tessl CLI
npx tessl i tessl/pypi-structlog