Observability API Reference

Structured logging, rich console output, and tracing utilities.

Overview

The dspu.observability module provides:

• Structured Logging: Context-based logging with ContextVar
• Rich Console Output: Colors, tables, trees, progress bars
• Tracing Decorators: @timed, @traced, @logged_errors
• Stream Capture: Redirect stdout/stderr to logger
• Configuration: JSON/YAML/HOCON config file support

Logging

Configuration

dspu.observability.logging.configure_logging

configure_logging(
    level: str | int = "INFO",
    format: str = "console",
    outputs: list[str] | None = None,
    show_time: bool = True,
    show_name: bool = True,
    show_level: bool = True,
) -> None

Configure logging with sensible defaults.

Parameters:

Name	Type	Description	Default
level	str | int	Log level (DEBUG, INFO, WARNING, ERROR, CRITICAL).	'INFO'
format	str	Output format ("console", "json", "rich").	'console'
outputs	list[str] | None	List of outputs (e.g., ["stdout", "file:app.log"]).	None
show_time	bool	Show timestamp in logs.	True
show_name	bool	Show logger name in logs.	True
show_level	bool	Show log level in logs.	True

Example

configure_logging(level="INFO", format="json") configure_logging(level="DEBUG", format="rich", outputs=["stdout", "file:app.log"])

StructuredLogger

dspu.observability.logging.get_logger

get_logger(name: str | None = None) -> StructuredLogger

Get a structured logger instance.

Parameters:

Name	Type	Description	Default
name	str | None	Logger name (typically name). If None, returns root logger.	None

Returns:

Type	Description
StructuredLogger	StructuredLogger instance.

Example

logger = get_logger(name) logger.info("Application started", version="1.0.0")

dspu.observability.logging.StructuredLogger

Bases: LoggerAdapter

Logger adapter that adds structured context to log records.

Example

logger = StructuredLogger(logging.getLogger(name)) logger.info("User login", user_id=123, ip="192.168.1.1")

Context Management

dspu.observability.logging.LogContext

LogContext(**context: Any)

Context manager for adding temporary log context.

Example

with LogContext(request_id="req-123", user_id=456): ... logger.info("Processing") # Includes request_id and user_id

Initialize log context.

Parameters:

Name	Type	Description	Default
**context	Any	Context fields to add.	{}

dspu.observability.logging.bind_context

bind_context(**context: Any) -> None

Add persistent context fields to all subsequent logs.

Parameters:

Name	Type	Description	Default
**context	Any	Context fields to add.	{}

Example

bind_context(service="api", version="1.0.0") logger.info("Started") # Includes service and version

Source code in src/dspu/observability/logging.py

def bind_context(**context: Any) -> None:
    """Add persistent context fields to all subsequent logs.

    Args:
        **context: Context fields to add.

    Example:
        >>> bind_context(service="api", version="1.0.0")
        >>> logger.info("Started")  # Includes service and version
    """
    current = (_log_context.get() or {}).copy()
    current.update(context)
    _log_context.set(current)

dspu.observability.logging.unbind_context

unbind_context(*keys: str) -> None

Remove context fields.

Parameters:

Name	Type	Description	Default
*keys	str	Context field keys to remove.	()

Example

unbind_context("request_id", "user_id")

Source code in src/dspu/observability/logging.py

def unbind_context(*keys: str) -> None:
    """Remove context fields.

    Args:
        *keys: Context field keys to remove.

    Example:
        >>> unbind_context("request_id", "user_id")
    """
    current = (_log_context.get() or {}).copy()
    for key in keys:
        current.pop(key, None)
    _log_context.set(current)

dspu.observability.logging.clear_context

clear_context() -> None

Clear all context fields.

Example

clear_context()

Source code in src/dspu/observability/logging.py

def clear_context() -> None:
    """Clear all context fields.

    Example:
        >>> clear_context()
    """
    _log_context.set({})

Decorators

Timing

dspu.observability.decorators.timed

timed(
    logger: Logger | str | None = None,
    level: int = INFO,
    threshold_ms: float | None = None,
) -> Callable[[F], F]

Decorator to log function execution time.

Parameters:

Name	Type	Description	Default
logger	Logger | str | None	Logger instance or name. If None, uses function's module logger.	None
level	int	Log level (default: INFO).	INFO
threshold_ms	float | None	Only log if execution time exceeds this threshold (milliseconds).	None

Returns:

Type	Description
Callable[[F], F]	Decorated function.

Example

@timed() ... def slow_function(): ... time.sleep(1) ... return "done" slow_function() # Logs: "slow_function took 1000.5ms"

@timed(threshold_ms=100) ... def fast_function(): ... return "quick" fast_function() # Not logged (below threshold)

Tracing

dspu.observability.decorators.traced

traced(
    logger: Logger | str | None = None,
    level: int = DEBUG,
    log_args: bool = True,
    log_result: bool = False,
) -> Callable[[F], F]

Decorator to trace function entry/exit and log arguments.

Parameters:

Name	Type	Description	Default
logger	Logger | str | None	Logger instance or name. If None, uses function's module logger.	None
level	int	Log level (default: DEBUG).	DEBUG
log_args	bool	If True, log function arguments.	True
log_result	bool	If True, log function return value.	False

Returns:

Type	Description
Callable[[F], F]	Decorated function.

Example

@traced() ... def process_order(order_id: str, items: list) -> dict: ... return {"status": "success"} process_order("123", ["item1", "item2"])

Logs: "Entering process_order(order_id='123', items=['item1', 'item2'])"

Logs: "Exiting process_order"

Error Logging

dspu.observability.decorators.logged_errors

logged_errors(
    logger: Logger | str | None = None,
    level: int = ERROR,
    reraise: bool = True,
    include_traceback: bool = True,
) -> Callable[[F], F]

Decorator to automatically log exceptions.

Parameters:

Name	Type	Description	Default
logger	Logger | str | None	Logger instance or name. If None, uses function's module logger.	None
level	int	Log level for errors (default: ERROR).	ERROR
reraise	bool	If True, re-raise exception after logging (default: True).	True
include_traceback	bool	If True, include full traceback in log.	True

Returns:

Type	Description
Callable[[F], F]	Decorated function.

Example

@logged_errors() ... def risky_operation(): ... raise ValueError("Something went wrong") risky_operation() # Logs error with traceback, then raises

@logged_errors(reraise=False) ... def safe_operation(): ... raise ValueError("Handled") ... return None safe_operation() # Logs error but doesn't raise

Rich Output

Console

dspu.observability.rich_output.get_console

get_console() -> Any

Get rich console instance.

Returns:

Type	Description
Any	Rich Console instance.

Raises:

Type	Description
ConfigurationError	If rich is not installed.

Example

console = get_console() console.print("[bold green]Success![/bold green]")

Source code in src/dspu/observability/rich_output.py

def get_console() -> Any:
    """Get rich console instance.

    Returns:
        Rich Console instance.

    Raises:
        ConfigurationError: If rich is not installed.

    Example:
        >>> console = get_console()
        >>> console.print("[bold green]Success![/bold green]")
    """
    global _console

    if _console is None:
        try:
            from rich.console import Console

            _console = Console()
        except ImportError as e:
            raise ConfigurationError(
                "Rich console requires rich library",
                suggestion="Install with: pip install rich",
            ) from e

    return _console

dspu.observability.rich_output.print_json

print_json(data: Any, **kwargs: Any) -> None

Pretty-print JSON data with syntax highlighting.

Parameters:

Name	Type	Description	Default
data	Any	Data to print (will be JSON-serialized).	required
**kwargs	Any	Additional arguments for rich JSON printing.	{}

Example

print_json({"user": "alice", "items": [1, 2, 3]})

Source code in src/dspu/observability/rich_output.py

def print_json(data: Any, **kwargs: Any) -> None:
    """Pretty-print JSON data with syntax highlighting.

    Args:
        data: Data to print (will be JSON-serialized).
        **kwargs: Additional arguments for rich JSON printing.

    Example:
        >>> print_json({"user": "alice", "items": [1, 2, 3]})
    """
    try:
        from rich.json import JSON
    except ImportError as e:
        raise ConfigurationError(
            "print_json requires rich library",
            suggestion="Install with: pip install rich",
        ) from e

    import json

    console = get_console()
    json_str = json.dumps(data, indent=2, default=str)
    console.print(JSON(json_str), **kwargs)

dspu.observability.rich_output.print_table

print_table(
    data: list[dict[str, Any]],
    title: str | None = None,
    **kwargs: Any,
) -> None

Display tabular data in a formatted table.

Parameters:

Name	Type	Description	Default
data	list[dict[str, Any]]	List of dictionaries to display.	required
title	str | None	Optional table title.	None
**kwargs	Any	Additional arguments for rich Table.	{}

Example

data = [ ... {"name": "Alice", "age": 30}, ... {"name": "Bob", "age": 25}, ... ] print_table(data, title="Users")

Source code in src/dspu/observability/rich_output.py

def print_table(
    data: list[dict[str, Any]],
    title: str | None = None,
    **kwargs: Any,
) -> None:
    """Display tabular data in a formatted table.

    Args:
        data: List of dictionaries to display.
        title: Optional table title.
        **kwargs: Additional arguments for rich Table.

    Example:
        >>> data = [
        ...     {"name": "Alice", "age": 30},
        ...     {"name": "Bob", "age": 25},
        ... ]
        >>> print_table(data, title="Users")
    """
    try:
        from rich.table import Table
    except ImportError as e:
        raise ConfigurationError(
            "print_table requires rich library",
            suggestion="Install with: pip install rich",
        ) from e

    if not data:
        return

    console = get_console()
    table = Table(title=title, **kwargs)

    # Add columns from first row
    for key in data[0].keys():
        table.add_column(str(key), style="cyan")

    # Add rows
    for row in data:
        table.add_row(*[str(v) for v in row.values()])

    console.print(table)

dspu.observability.rich_output.print_tree

print_tree(
    data: dict[str, Any], title: str = "Data"
) -> None

Display hierarchical data as a tree.

Parameters:

Name	Type	Description	Default
data	dict[str, Any]	Nested dictionary to display.	required
title	str	Tree title.	'Data'

Example

data = { ... "root": { ... "child1": {"leaf1": "value1"}, ... "child2": {"leaf2": "value2"}, ... } ... } print_tree(data)

Source code in src/dspu/observability/rich_output.py

def print_tree(data: dict[str, Any], title: str = "Data") -> None:
    """Display hierarchical data as a tree.

    Args:
        data: Nested dictionary to display.
        title: Tree title.

    Example:
        >>> data = {
        ...     "root": {
        ...         "child1": {"leaf1": "value1"},
        ...         "child2": {"leaf2": "value2"},
        ...     }
        ... }
        >>> print_tree(data)
    """
    try:
        from rich.tree import Tree
    except ImportError as e:
        raise ConfigurationError(
            "print_tree requires rich library",
            suggestion="Install with: pip install rich",
        ) from e

    console = get_console()

    def build_tree(node: Tree, data: Any) -> None:
        """Recursively build tree."""
        if isinstance(data, dict):
            for key, value in data.items():
                if isinstance(value, (dict, list)):
                    branch = node.add(f"[bold]{key}[/bold]")
                    build_tree(branch, value)
                else:
                    node.add(f"{key}: {value}")
        elif isinstance(data, list):
            for i, item in enumerate(data):
                if isinstance(item, (dict, list)):
                    branch = node.add(f"[bold][{i}][/bold]")
                    build_tree(branch, item)
                else:
                    node.add(f"[{i}]: {item}")

    tree = Tree(title)
    build_tree(tree, data)
    console.print(tree)

dspu.observability.rich_output.print_traceback

print_traceback(
    *,
    show_locals: bool = False,
    max_frames: int = 20,
    **kwargs: Any,
) -> None

Print enhanced traceback for current exception.

Parameters:

Name	Type	Description	Default
show_locals	bool	Show local variables in traceback.	False
max_frames	int	Maximum number of stack frames to show.	20
**kwargs	Any	Additional arguments for rich Traceback.	{}

Example

try: ... risky_operation() ... except Exception: ... print_traceback(show_locals=True)

Source code in src/dspu/observability/rich_output.py

def print_traceback(
    *,
    show_locals: bool = False,
    max_frames: int = 20,
    **kwargs: Any,
) -> None:
    """Print enhanced traceback for current exception.

    Args:
        show_locals: Show local variables in traceback.
        max_frames: Maximum number of stack frames to show.
        **kwargs: Additional arguments for rich Traceback.

    Example:
        >>> try:
        ...     risky_operation()
        ... except Exception:
        ...     print_traceback(show_locals=True)
    """
    try:
        from rich.traceback import Traceback
    except ImportError as e:
        raise ConfigurationError(
            "print_traceback requires rich library",
            suggestion="Install with: pip install rich",
        ) from e

    console = get_console()
    console.print(
        Traceback(
            show_locals=show_locals,
            max_frames=max_frames,
            **kwargs,
        )
    )

dspu.observability.rich_output.inspect_object

inspect_object(
    obj: Any,
    methods: bool = False,
    private: bool = False,
    **kwargs: Any,
) -> None

Rich inspection of an object.

Parameters:

Name	Type	Description	Default
obj	Any	Object to inspect.	required
methods	bool	Show methods.	False
private	bool	Show private attributes.	False
**kwargs	Any	Additional arguments for rich inspect.	{}

Example

inspect_object(my_object, methods=True)

Source code in src/dspu/observability/rich_output.py

def inspect_object(
    obj: Any,
    methods: bool = False,
    private: bool = False,
    **kwargs: Any,
) -> None:
    """Rich inspection of an object.

    Args:
        obj: Object to inspect.
        methods: Show methods.
        private: Show private attributes.
        **kwargs: Additional arguments for rich inspect.

    Example:
        >>> inspect_object(my_object, methods=True)
    """
    try:
        from rich import inspect as rich_inspect
    except ImportError as e:
        raise ConfigurationError(
            "inspect_object requires rich library",
            suggestion="Install with: pip install rich",
        ) from e

    rich_inspect(
        obj,
        methods=methods,
        private=private,
        console=get_console(),
        **kwargs,
    )

Progress

dspu.observability.rich_output.Progress

Progress(**kwargs: Any)

Progress bar context manager.

Example

with Progress() as progress: ... task = progress.add_task("Processing", total=1000) ... for i in range(1000): ... process_item(i) ... progress.update(task, advance=1)

Initialize progress.

Parameters:

Name	Type	Description	Default
**kwargs	Any	Arguments for rich Progress.	{}

Source code in src/dspu/observability/rich_output.py

def __init__(self, **kwargs: Any):
    """Initialize progress.

    Args:
        **kwargs: Arguments for rich Progress.
    """
    try:
        from rich.progress import Progress as RichProgress

        self._progress_class = RichProgress
    except ImportError as e:
        raise ConfigurationError(
            "Progress requires rich library",
            suggestion="Install with: pip install rich",
        ) from e

    self._progress = None
    self._kwargs = kwargs

dspu.observability.rich_output.TaskProgress

TaskProgress(**kwargs: Any)

Multi-task progress tracking.

Example

with TaskProgress() as progress: ... download = progress.add_task("Downloading", total=100) ... extract = progress.add_task("Extracting", total=100) ... for i in range(100): ... progress.update(download, advance=1)

Initialize task progress.

Parameters:

Name	Type	Description	Default
**kwargs	Any	Arguments for rich Progress.	{}

Source code in src/dspu/observability/rich_output.py

def __init__(self, **kwargs: Any):
    """Initialize task progress.

    Args:
        **kwargs: Arguments for rich Progress.
    """
    try:
        from rich.progress import (
            BarColumn,
            MofNCompleteColumn,
            SpinnerColumn,
            TaskProgressColumn,
            TextColumn,
            TimeElapsedColumn,
        )
        from rich.progress import (
            Progress as RichProgress,
        )

        self._progress_class = RichProgress
        self._columns = [
            SpinnerColumn(),
            TextColumn("[progress.description]{task.description}"),
            BarColumn(),
            TaskProgressColumn(),
            MofNCompleteColumn(),
            TimeElapsedColumn(),
        ]
    except ImportError as e:
        raise ConfigurationError(
            "TaskProgress requires rich library",
            suggestion="Install with: pip install rich",
        ) from e

    self._progress = None
    self._kwargs = kwargs

Advanced

LoggingSetup

dspu.observability.setup.LoggingSetup

LoggingSetup(
    default_log_config_dir: str = ".",
    default_log_config_file: str = "logging.json",
    default_log_level: str = "DEBUG",
    log_config_env: str | None = None,
    log_level_env: str | None = None,
    log_file: str | None = None,
    logging_config: dict[str, Any] | str | None = None,
    source_anchor: Any = None,
    default_logger_name: str | None = None,
)

Configure Python logging from files, dictionaries, or environment variables.

Provides flexible logging configuration with support for: - Configuration from JSON, YAML, or HOCON files - Environment variable overrides - Dictionary configuration - Relative path resolution - Stdout/stderr redirection

Example

setup = LoggingSetup( ... default_log_config_file="logging.json", ... default_log_level="INFO", ... log_config_env="LOG_CONFIG", ... log_level_env="LOG_LEVEL", ... ) setup.setup()

Initialize logging setup.

Parameters:

Name	Type	Description	Default
default_log_config_dir	str	Directory where logging config file lives.	'.'
default_log_config_file	str	Default name of logging config file.	'logging.json'
default_log_level	str	Default log level if no config file found.	'DEBUG'
log_config_env	str | None	Environment variable for config file override.	None
log_level_env	str | None	Environment variable for log level override.	None
log_file	str | None	Actual file to log to (overrides config).	None
logging_config	dict[str, Any] | str | None	Config reference (file path or dictionary).	None
source_anchor	Any	Object whose source file is anchor for relative paths.	None
default_logger_name	str | None	Logger name for default setup.	None

Source code in src/dspu/observability/setup.py

def __init__(
    self,
    default_log_config_dir: str = ".",
    default_log_config_file: str = "logging.json",
    default_log_level: str = "DEBUG",
    log_config_env: str | None = None,
    log_level_env: str | None = None,
    log_file: str | None = None,
    logging_config: dict[str, Any] | str | None = None,
    source_anchor: Any = None,
    default_logger_name: str | None = None,
):
    """Initialize logging setup.

    Args:
        default_log_config_dir: Directory where logging config file lives.
        default_log_config_file: Default name of logging config file.
        default_log_level: Default log level if no config file found.
        log_config_env: Environment variable for config file override.
        log_level_env: Environment variable for log level override.
        log_file: Actual file to log to (overrides config).
        logging_config: Config reference (file path or dictionary).
        source_anchor: Object whose source file is anchor for relative paths.
        default_logger_name: Logger name for default setup.
    """
    self.default_log_config_dir = default_log_config_dir
    self.default_log_config_file = default_log_config_file
    self.default_log_level = default_log_level
    self.log_config_env = log_config_env
    self.log_level_env = log_level_env
    self.log_file = log_file
    self.logging_config = logging_config
    self.source_anchor = source_anchor
    self.default_logger_name = default_logger_name

Functions

setup

setup() -> None

Set up logging per the configured parameters.

Configures Python logging based on: 1. Explicit logging_config (if provided) 2. Config file path from environment variable or defaults 3. Basic config with default log level

Raises:

Type	Description
ConfigurationError	If config file cannot be loaded.

Source code in src/dspu/observability/setup.py

def setup(self) -> None:
    """Set up logging per the configured parameters.

    Configures Python logging based on:
    1. Explicit logging_config (if provided)
    2. Config file path from environment variable or defaults
    3. Basic config with default log level

    Raises:
        ConfigurationError: If config file cannot be loaded.
    """
    # First, assume config is what we got from constructor
    config = self.logging_config

    # Determine config file path if needed
    log_config_file_path = self.determine_log_config_file_path()
    if log_config_file_path is not None:
        # Read the logging config file
        config = self._load_config_file(log_config_file_path)

    # Apply the configuration
    if config is not None and isinstance(config, dict):
        config = self.replace_log_file(config)
        logging.config.dictConfig(config)
    else:
        # Fall back to basic config
        log_level = self.determine_log_level()
        logging.basicConfig(
            filename=self.log_file,
            level=log_level,
            format="%(asctime)s - %(name)s - %(levelname)s - %(message)s",
        )

setup_with_diversion

setup_with_diversion() -> Logger

Set up logging and divert stdout/stderr to logger.

Returns:

Type	Description
Logger	Logger instance used for diversion.

Example

logger = setup.setup_with_diversion() print("This goes to the logger")

Source code in src/dspu/observability/setup.py

def setup_with_diversion(self) -> logging.Logger:
    """Set up logging and divert stdout/stderr to logger.

    Returns:
        Logger instance used for diversion.

    Example:
        >>> logger = setup.setup_with_diversion()
        >>> print("This goes to the logger")
    """
    self.setup()

    # Determine the default logger name
    logger_name = self.default_logger_name
    if logger_name is None:
        logger_name = "default"
        if self.source_anchor is not None:
            logger_name = self.source_anchor.__class__.__name__

    logger = logging.getLogger(logger_name)
    log_level = self.determine_log_level()
    logger.setLevel(log_level)

    # Import StreamToLogger to avoid circular import
    from dspu.observability.stream_capture import StreamToLogger

    StreamToLogger.subvert(
        logger=logger,
        reroute_stdout=True,
        reroute_stderr=True,
    )

    logger.info("Print statements diverted to logger")
    return logger

Stream Capture

dspu.observability.stream_capture.StreamToLogger

StreamToLogger(logger: Logger, log_level: int = INFO)

Bases: StringIO

File-like stream that redirects writes to a logger.

Redirects stdout/stderr to a logger instance, useful for capturing print statements in logging output. Includes protection against infinite recursion when loggers output to stdout.

Example

logger = logging.getLogger(name) StreamToLogger.subvert(logger, reroute_stdout=True) print("This goes to the logger")

Initialize stream-to-logger adapter.

Parameters:

Name	Type	Description	Default
logger	Logger	Logger to redirect writes to.	required
log_level	int	Logging level for writes (default: INFO).	INFO

Source code in src/dspu/observability/stream_capture.py

def __init__(self, logger: logging.Logger, log_level: int = logging.INFO):
    """Initialize stream-to-logger adapter.

    Args:
        logger: Logger to redirect writes to.
        log_level: Logging level for writes (default: INFO).
    """
    super().__init__()
    self.logger = logger
    self.log_level = log_level

    # Stacks to help avoid infinite recursion
    self.last_logged: list[str] = []
    self.last_num_lines_logged: list[int] = []

Functions

subvert classmethod

subvert(
    logger: Logger | None = None,
    reroute_stdout: bool = True,
    reroute_stderr: bool = True,
) -> None

Subvert stdout and stderr to logger.

Parameters:

Name	Type	Description	Default
logger	Logger | None	Logger to use for subverting stdout/stderr. If None, creates new logger from class name.	None
reroute_stdout	bool	If True, reroute stdout to logger.	True
reroute_stderr	bool	If True, reroute stderr to logger.	True

Example

logger = logging.getLogger(name) StreamToLogger.subvert(logger, reroute_stdout=True) print("This goes to the logger at INFO level")

Source code in src/dspu/observability/stream_capture.py

@classmethod
def subvert(
    cls,
    logger: logging.Logger | None = None,
    reroute_stdout: bool = True,
    reroute_stderr: bool = True,
) -> None:
    """Subvert stdout and stderr to logger.

    Args:
        logger: Logger to use for subverting stdout/stderr.
               If None, creates new logger from class name.
        reroute_stdout: If True, reroute stdout to logger.
        reroute_stderr: If True, reroute stderr to logger.

    Example:
        >>> logger = logging.getLogger(__name__)
        >>> StreamToLogger.subvert(logger, reroute_stdout=True)
        >>> print("This goes to the logger at INFO level")
    """
    # Get or create logger
    use_logger = logger
    if use_logger is None:
        use_logger = logging.getLogger(cls.__name__)

    # Reroute stdout to INFO level
    if reroute_stdout:
        info_stream_logger = cls(use_logger, logging.INFO)
        sys.stdout = info_stream_logger  # type: ignore

    # Reroute stderr to ERROR level
    if reroute_stderr:
        error_stream_logger = cls(use_logger, logging.ERROR)
        sys.stderr = error_stream_logger  # type: ignore

restore classmethod

restore() -> None

Restore stdout and stderr to their original state.

Example

StreamToLogger.restore() print("This goes to normal stdout")

Source code in src/dspu/observability/stream_capture.py

@classmethod
def restore(cls) -> None:
    """Restore stdout and stderr to their original state.

    Example:
        >>> StreamToLogger.restore()
        >>> print("This goes to normal stdout")
    """
    sys.stdout = sys.__stdout__
    sys.stderr = sys.__stderr__

Usage Examples

Basic Logging

from dspu.observability import configure_logging, get_logger, LogContext

# Configure
configure_logging(level="INFO", format="rich")
logger = get_logger(__name__)

# Log with structured fields
logger.info("User logged in", user_id=123, ip="192.168.1.1")

# Context manager
with LogContext(request_id="req-123"):
    logger.info("Processing request")  # Includes request_id
    logger.info("Query executed", duration_ms=45)

Tracing Decorators

from dspu.observability import timed, traced, logged_errors

@timed()
def slow_operation():
    # Execution time automatically logged
    time.sleep(1)

@traced(log_args=True, log_result=True)
def process_order(order_id, items):
    # Logs entry, args, result, and exit
    return {"status": "success", "order_id": order_id}

@logged_errors(reraise=True)
def risky_operation():
    # Logs errors with traceback
    raise ValueError("Something went wrong")

Rich Output

from dspu.observability import print_json, print_table, Progress

# Pretty JSON
data = {"user": {"name": "Alice", "age": 30}}
print_json(data)

# Table
users = [{"name": "Alice", "age": 30}, {"name": "Bob", "age": 25}]
print_table(users, title="Users")

# Progress bar
with Progress() as progress:
    task = progress.add_task("Processing", total=100)
    for i in range(100):
        time.sleep(0.01)
        progress.update(task, advance=1)

See Also

• Observability Examples - Practical examples
• Observability User Guide - Detailed guide
• Quick Start - Get started quickly