tessl/pypi-structlog

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

Overview

Eval results

Files

Context Management

Name: tessl/pypi-structlog
Author: tessl

Thread-safe context binding using contextvars and deprecated thread-local storage, enabling global context that persists across function calls and async boundaries. Context management allows you to maintain structured data that automatically appears in all log entries within a given scope.

Capabilities

Context Variables (Recommended)

Modern context management using Python's contextvars module for thread-safe and async-safe context storage.

def get_contextvars() -> dict[str, Any]:
    """
    Return copy of structlog-specific context-local context.
    
    Returns:
        dict: Current context variables as key-value pairs
    """

def get_merged_contextvars(bound_logger) -> dict[str, Any]:
    """
    Return merged context-local and bound logger context.
    
    Args:
        bound_logger: Bound logger instance
        
    Returns:
        dict: Merged context from contextvars and bound logger
    """

def merge_contextvars(logger, method_name, event_dict) -> EventDict:
    """
    Processor that merges global context-local context into event dict.
    
    Use this processor to automatically include context variables
    in all log entries.
    
    Args:
        logger: Logger instance
        method_name (str): Logger method name
        event_dict (dict): Event dictionary
        
    Returns:
        dict: Event dictionary with context variables merged in
    """

def clear_contextvars() -> None:
    """Clear all structlog context-local context variables."""

def bind_contextvars(**kw) -> Mapping[str, contextvars.Token[Any]]:
    """
    Put keys and values into context-local context.
    
    Args:
        **kw: Key-value pairs to bind to context
        
    Returns:
        dict: Mapping of keys to context variable tokens for later reset
    """

def reset_contextvars(**kw) -> None:
    """
    Reset contextvars using tokens returned from bind_contextvars.
    
    Args:
        **kw: Mapping of keys to tokens from previous bind_contextvars call
    """

def unbind_contextvars(*keys) -> None:
    """
    Remove keys from context-local context.
    
    Args:
        *keys: Context variable keys to remove
    """

def bound_contextvars(**kw) -> Generator[None, None, None]:
    """
    Context manager for temporary context binding.
    
    Binds context variables for the duration of the with block,
    then automatically restores the previous state.
    
    Args:
        **kw: Key-value pairs to bind temporarily
        
    Yields:
        None
    """

Context Variable Constants

STRUCTLOG_KEY_PREFIX: str
"""Prefix used for structlog context variable names."""

Thread-Local Context (Deprecated)

Legacy thread-local context management. Use contextvars instead for new code.

def wrap_dict(dict_class) -> type[Context]:
    """
    Wrap a dict class to be used for thread-local storage.
    
    Args:
        dict_class: Dictionary class to wrap
        
    Returns:
        type: Wrapped dictionary class with thread-local behavior
    """

def as_immutable(logger) -> logger:
    """
    Extract context from thread-local storage into an immutable logger.
    
    Args:
        logger: Logger instance
        
    Returns:
        Logger with context extracted from thread-local storage
    """

def tmp_bind(logger, **tmp_values) -> Generator[logger, None, None]:
    """
    Context manager for temporarily binding thread-local context.
    
    Args:
        logger: Logger instance
        **tmp_values: Temporary values to bind
        
    Yields:
        Logger: Logger with temporary context bound
    """

def get_threadlocal() -> Context:
    """Get copy of current thread-local context."""

def get_merged_threadlocal(bound_logger) -> Context:
    """
    Get merged thread-local and bound logger context.
    
    Args:
        bound_logger: Bound logger instance
        
    Returns:
        dict: Merged context
    """

def merge_threadlocal(logger, method_name, event_dict) -> EventDict:
    """
    Processor for merging thread-local context into event dict.
    
    Args:
        logger: Logger instance
        method_name (str): Logger method name
        event_dict (dict): Event dictionary
        
    Returns:
        dict: Event dictionary with thread-local context merged
    """

def clear_threadlocal() -> None:
    """Clear all thread-local context."""

def bind_threadlocal(**kw) -> None:
    """
    Bind key-value pairs to thread-local context.
    
    Args:
        **kw: Key-value pairs to bind
    """

def unbind_threadlocal(*keys) -> None:
    """
    Remove keys from thread-local context.
    
    Args:
        *keys: Keys to remove
    """

def bound_threadlocal(**kw) -> Generator[None, None, None]:
    """
    Context manager for temporary thread-local context binding.
    
    Args:
        **kw: Key-value pairs to bind temporarily
        
    Yields:
        None
    """

Usage Examples

Basic Context Variables Usage

import structlog
from structlog import contextvars, processors

# Configure structlog with context variables
structlog.configure(
    processors=[
        contextvars.merge_contextvars,  # Include context vars in logs
        processors.TimeStamper(),
        processors.JSONRenderer()
    ],
    wrapper_class=structlog.BoundLogger,
)

logger = structlog.get_logger()

# Bind context variables globally
contextvars.bind_contextvars(
    service="api",
    version="1.0.0"
)

# All subsequent logs include the context
logger.info("Service started")  # Includes service and version
logger.info("Processing request", request_id="req-123")

Temporary Context Binding

import structlog
from structlog import contextvars

structlog.configure(
    processors=[
        contextvars.merge_contextvars,
        structlog.processors.JSONRenderer()
    ],
    wrapper_class=structlog.BoundLogger,
)

logger = structlog.get_logger()

# Set some global context
contextvars.bind_contextvars(service="user-service")

# Temporarily add context for a block
with contextvars.bound_contextvars(user_id=123, operation="update"):
    logger.info("Starting user update")  # Includes user_id and operation
    logger.info("Validation passed")     # Includes user_id and operation

# Outside the block, temporary context is gone
logger.info("Update completed")  # Only includes service

Request-Scoped Context

import structlog
from structlog import contextvars
import uuid

def process_request(request_data):
    """Process a request with request-scoped context."""
    
    # Generate request ID and bind context
    request_id = str(uuid.uuid4())
    
    with contextvars.bound_contextvars(
        request_id=request_id,
        user_id=request_data.get("user_id"),
        endpoint=request_data.get("endpoint")
    ):
        logger = structlog.get_logger()
        
        logger.info("Request started")
        
        # All nested function calls will include the context
        validate_request(request_data)
        result = perform_operation(request_data)
        
        logger.info("Request completed", result_count=len(result))
        return result

def validate_request(data):
    """Validation function - automatically gets request context."""
    logger = structlog.get_logger()
    logger.info("Validating request", fields=list(data.keys()))

def perform_operation(data):
    """Business logic - automatically gets request context."""
    logger = structlog.get_logger()
    logger.info("Performing operation", operation_type=data.get("type"))
    return ["result1", "result2"]

Context Management Across Async Calls

import asyncio
import structlog
from structlog import contextvars

async def handle_async_request(user_id, task_type):
    """Handle async request with context that spans async boundaries."""
    
    # Bind context for this async task
    with contextvars.bound_contextvars(
        user_id=user_id,
        task_type=task_type,
        correlation_id=str(uuid.uuid4())
    ):
        logger = structlog.get_logger()
        
        logger.info("Async task started")
        
        # Context is preserved across await calls
        result = await perform_async_operation()
        
        logger.info("Async task completed", result=result)
        return result

async def perform_async_operation():
    """Async operation that inherits context."""
    logger = structlog.get_logger()
    
    logger.info("Starting async operation")
    await asyncio.sleep(0.1)  # Simulate async work
    logger.info("Async operation completed")
    
    return "success"

# Usage
async def main():
    await handle_async_request(user_id=123, task_type="data_processing")

asyncio.run(main())

Context Inspection and Management

import structlog
from structlog import contextvars

structlog.configure(
    processors=[
        contextvars.merge_contextvars,
        structlog.processors.JSONRenderer()
    ],
    wrapper_class=structlog.BoundLogger,
)

# Bind some context
contextvars.bind_contextvars(
    service="auth",
    environment="production",
    debug_mode=False
)

# Inspect current context
current_context = contextvars.get_contextvars()
print(f"Current context: {current_context}")

# Conditionally add more context
if current_context.get("environment") == "production":
    contextvars.bind_contextvars(monitoring_enabled=True)

# Remove specific context
contextvars.unbind_contextvars("debug_mode")

# Clear all context
# contextvars.clear_contextvars()

Token-Based Context Reset

import structlog
from structlog import contextvars

# Bind context and get tokens
tokens = contextvars.bind_contextvars(
    transaction_id="tx-123",
    user_id=456
)

logger = structlog.get_logger()
logger.info("Transaction started")

# Later, reset specific context using tokens
contextvars.reset_contextvars(user_id=tokens["user_id"])

logger.info("User context reset")

Mixed Context Sources

import structlog
from structlog import contextvars

structlog.configure(
    processors=[
        contextvars.merge_contextvars,
        structlog.processors.JSONRenderer()
    ],
    wrapper_class=structlog.BoundLogger,
)

# Global context via contextvars
contextvars.bind_contextvars(service="api", version="2.0")

# Logger-specific context via bind()
logger = structlog.get_logger().bind(component="auth")

# Both contexts will be merged
logger.info("Authentication attempt", username="alice")
# Output includes: service, version, component, and username

# Get merged context from logger
bound_logger = structlog.get_logger().bind(session="abc-123")
merged = contextvars.get_merged_contextvars(bound_logger)
print(f"Merged context: {merged}")

Error Handling with Context

import structlog
from structlog import contextvars

def risky_operation():
    """Operation that might fail - context helps with debugging."""
    
    with contextvars.bound_contextvars(
        operation="data_processing",
        batch_id="batch-456"
    ):
        logger = structlog.get_logger()
        
        try:
            logger.info("Starting risky operation")
            
            # Simulate error
            raise ValueError("Something went wrong")
            
        except Exception as e:
            logger.error(
                "Operation failed",
                error_type=type(e).__name__,
                error_message=str(e)
            )
            # Context variables (operation, batch_id) automatically included
            raise

# The error log will include operation and batch_id for easier debugging

Install with Tessl CLI