tessl/pypi-structlog

Structured logging for Python that emphasizes simplicity, power, and performance

Overview

Eval results

Files

Log Processors

Name: tessl/pypi-structlog
Author: tessl

Extensible pipeline of functions that transform, filter, and format log events as they flow through the structlog system. Processors can add data, remove data, change formats, or perform side effects like writing to files or sending to external systems.

Capabilities

Timestamp Processors

Add timestamp information to log events with configurable formatting and timezone handling.

class TimeStamper:
    """Add timestamp to event dictionary."""
    
    def __init__(self, fmt=None, utc=True, key="timestamp"):
        """
        Args:
            fmt (str, optional): Timestamp format ("iso", datetime format string, or None for epoch)
            utc (bool): Use UTC timezone (default: True)
            key (str): Key name for timestamp in event dict (default: "timestamp")
        """
    
    def __call__(self, logger, name, event_dict) -> EventDict: ...

class MaybeTimeStamper:
    """Add timestamp only if the timestamp key doesn't already exist."""
    
    def __init__(self, fmt=None, utc=True, key="timestamp"):
        """
        Args:
            fmt (str, optional): Timestamp format
            utc (bool): Use UTC timezone
            key (str): Key name for timestamp
        """
    
    def __call__(self, logger, name, event_dict) -> EventDict: ...

Rendering Processors

Transform structured event dictionaries into formatted strings for output.

class JSONRenderer:
    """Render event dictionary as JSON string."""
    
    def __init__(self, serializer=json.dumps, **dumps_kw):
        """
        Args:
            serializer (callable): JSON serialization function (default: json.dumps)
            **dumps_kw: Additional keyword arguments for serializer
        """
    
    def __call__(self, logger, name, event_dict) -> str | bytes: ...

class KeyValueRenderer:
    """Render event dictionary as key=value pairs."""
    
    def __init__(self, sort_keys=False, key_order=None, drop_missing=False, repr_native_str=True):
        """
        Args:
            sort_keys (bool): Sort keys alphabetically
            key_order (list, optional): Specific key ordering
            drop_missing (bool): Drop keys with None values
            repr_native_str (bool): Use repr() for string values
        """
    
    def __call__(self, logger, method_name, event_dict) -> str: ...

class LogfmtRenderer:
    """Render event dictionary in logfmt format."""
    
    def __init__(self, sort_keys=False, key_order=None, drop_missing=False, bool_as_flag=True):
        """
        Args:
            sort_keys (bool): Sort keys alphabetically
            key_order (list, optional): Specific key ordering
            drop_missing (bool): Drop keys with None values
            bool_as_flag (bool): Render True as flag, False as key=false
        """
    
    def __call__(self, logger, method_name, event_dict) -> str: ...

Text Processing Processors

Handle text encoding, decoding, and formatting operations.

class UnicodeEncoder:
    """Encode unicode values in event dictionary."""
    
    def __init__(self, encoding="utf-8", errors="backslashreplace"):
        """
        Args:
            encoding (str): Text encoding to use
            errors (str): Error handling strategy
        """
    
    def __call__(self, logger, name, event_dict) -> EventDict: ...

class UnicodeDecoder:
    """Decode byte string values in event dictionary."""
    
    def __init__(self, encoding="utf-8", errors="replace"):
        """
        Args:
            encoding (str): Text encoding to use
            errors (str): Error handling strategy for decode errors
        """
    
    def __call__(self, logger, name, event_dict) -> EventDict: ...

Exception Handling Processors

Process and format exception information in log events.

class ExceptionRenderer:
    """Replace exc_info tuple with formatted exception string."""
    
    def __init__(self, exception_formatter=_format_exception):
        """
        Args:
            exception_formatter (callable): Function to format exceptions
        """
    
    def __call__(self, logger, name, event_dict) -> EventDict: ...

class ExceptionPrettyPrinter:
    """Pretty print exceptions and remove exc_info from event dict."""
    
    def __init__(self, file=None, exception_formatter=_format_exception):
        """
        Args:
            file (file-like, optional): File to write exceptions to
            exception_formatter (callable): Function to format exceptions
        """
    
    def __call__(self, logger, name, event_dict) -> EventDict: ...

Stack Information Processors

Add stack frame and call site information to log events.

class StackInfoRenderer:
    """Add stack information to event dictionary."""
    
    def __init__(self, additional_ignores=None):
        """
        Args:
            additional_ignores (list, optional): Additional modules to ignore in stack traces
        """
    
    def __call__(self, logger, name, event_dict) -> EventDict: ...

class CallsiteParameterAdder:
    """Add call site parameters (filename, line number, etc.) to event dictionary."""
    
    def __init__(self, parameters=CallsiteParameter.all_parameters, additional_ignores=None):
        """
        Args:
            parameters (list): List of CallsiteParameter enum values to include
            additional_ignores (list, optional): Additional modules to ignore
        """
    
    def __call__(self, logger, name, event_dict) -> EventDict: ...

class CallsiteParameter(Enum):
    """Enumeration of available call site parameters."""
    PATHNAME = "pathname"
    FILENAME = "filename"
    MODULE = "module"
    FUNC_NAME = "funcName"
    LINENO = "lineno"
    THREAD = "thread"
    THREAD_NAME = "threadName"
    PROCESS = "process"
    PROCESS_NAME = "processName"

Utility Processors

General-purpose processors for common transformations and operations.

class EventRenamer:
    """Rename the event key to a different name."""
    
    def __init__(self, to: str, replace_by=None):
        """
        Args:
            to (str): New name for the event key
            replace_by (str, optional): Replace event value if key already exists
        """
    
    def __call__(self, logger, name, event_dict) -> EventDict: ...

def add_log_level(logger, method_name, event_dict) -> EventDict:
    """
    Add log level name to event dictionary.
    
    Args:
        logger: Logger instance
        method_name (str): Logger method name
        event_dict (dict): Event dictionary
        
    Returns:
        dict: Event dictionary with 'level' key added
    """

Pre-built Processor Instances

Common processor instances ready for use in configuration.

format_exc_info: ExceptionRenderer
"""Pre-configured ExceptionRenderer instance for formatting exceptions."""

dict_tracebacks: ExceptionRenderer  
"""Pre-configured ExceptionRenderer instance for structured traceback dictionaries."""

Usage Examples

Basic Processor Chain

import structlog
from structlog import processors

structlog.configure(
    processors=[
        processors.TimeStamper(fmt="iso"),
        processors.add_log_level,
        processors.JSONRenderer()
    ],
    wrapper_class=structlog.BoundLogger,
)

logger = structlog.get_logger()
logger.info("Test message")
# Output: {"timestamp": "2023-10-01T12:00:00Z", "level": "info", "event": "Test message"}

Custom Formatting Chain

import structlog
from structlog import processors

structlog.configure(
    processors=[
        processors.TimeStamper(fmt="%Y-%m-%d %H:%M:%S", utc=False),
        processors.add_log_level,
        processors.KeyValueRenderer(sort_keys=True, key_order=["timestamp", "level", "event"])
    ],
    wrapper_class=structlog.BoundLogger,
)

logger = structlog.get_logger()
logger.info("Test message", user_id=123)
# Output: timestamp='2023-10-01 12:00:00' level='info' event='Test message' user_id=123

Exception Processing

import structlog
from structlog import processors

structlog.configure(
    processors=[
        processors.TimeStamper(),
        processors.ExceptionRenderer(),
        processors.JSONRenderer()
    ],
    wrapper_class=structlog.BoundLogger,
)

logger = structlog.get_logger()

try:
    1 / 0
except ZeroDivisionError:
    logger.exception("Division by zero occurred")
    # Exception information will be formatted and included in JSON output

Stack Information

import structlog
from structlog import processors

structlog.configure(
    processors=[
        processors.CallsiteParameterAdder(
            parameters=[
                processors.CallsiteParameter.FILENAME,
                processors.CallsiteParameter.LINENO,
                processors.CallsiteParameter.FUNC_NAME
            ]
        ),
        processors.JSONRenderer()
    ],
    wrapper_class=structlog.BoundLogger,
)

logger = structlog.get_logger()
logger.info("Debug message")
# Output includes filename, line number, and function name

Text Processing

import structlog
from structlog import processors

structlog.configure(
    processors=[
        processors.UnicodeEncoder(encoding="utf-8"),
        processors.KeyValueRenderer()
    ],
    wrapper_class=structlog.BoundLogger,
)

logger = structlog.get_logger()
logger.info("Message with unicode", text="Hello 世界")

Custom Processor

import structlog
from structlog import processors

def add_hostname_processor(logger, method_name, event_dict):
    """Custom processor to add hostname to every log entry."""
    import socket
    event_dict["hostname"] = socket.gethostname()
    return event_dict

structlog.configure(
    processors=[
        processors.TimeStamper(),
        add_hostname_processor,
        processors.JSONRenderer()
    ],
    wrapper_class=structlog.BoundLogger,
)

logger = structlog.get_logger()
logger.info("Test message")
# Output includes hostname field

Conditional Processing

import structlog
from structlog import processors

def sensitive_data_filter(logger, method_name, event_dict):
    """Remove sensitive data from logs."""
    sensitive_keys = ["password", "token", "secret"]
    for key in sensitive_keys:
        if key in event_dict:
            event_dict[key] = "[REDACTED]"
    return event_dict

structlog.configure(
    processors=[
        sensitive_data_filter,
        processors.TimeStamper(),
        processors.JSONRenderer()
    ],
    wrapper_class=structlog.BoundLogger,
)

logger = structlog.get_logger()
logger.info("User login", username="alice", password="secret123")
# Output: password field will show "[REDACTED]"

Processor with Configuration

import structlog
from structlog import processors

# Configure timestamp processor
timestamper = processors.TimeStamper(
    fmt="iso",
    utc=True,
    key="@timestamp"  # Use custom key for timestamp
)

# Configure JSON renderer with custom options
json_renderer = processors.JSONRenderer(
    sort_keys=True,
    ensure_ascii=False,
    separators=(',', ':')
)

structlog.configure(
    processors=[
        timestamper,
        processors.add_log_level,
        json_renderer
    ],
    wrapper_class=structlog.BoundLogger,
)

Install with Tessl CLI