tessl/pypi-structlog

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

Overview

Eval results

Files

Output Loggers

Name: tessl/pypi-structlog
Author: tessl

Direct file output loggers that bypass standard logging infrastructure for high-performance logging scenarios and simple output requirements. These loggers write directly to files or streams without the overhead of Python's standard logging module.

Capabilities

Print Logger

Logger that prints messages to a file using Python's print() function.

class PrintLogger:
    """
    Logger that prints log messages to a file.
    
    Uses Python's print() function to write log messages,
    making it simple but potentially slower than WriteLogger.
    """
    
    def __init__(self, file=None):
        """
        Initialize PrintLogger.
        
        Args:
            file (file-like, optional): File object to print to.
                                       Defaults to sys.stdout if None.
        """
    
    def msg(self, message: str) -> None:
        """
        Print a log message.
        
        Args:
            message (str): Message to print
        """
    
    # Alias methods for different log levels
    def log(self, message: str) -> None: ...
    def debug(self, message: str) -> None: ...
    def info(self, message: str) -> None: ...
    def warn(self, message: str) -> None: ...
    def warning(self, message: str) -> None: ...
    def err(self, message: str) -> None: ...
    def error(self, message: str) -> None: ...
    def critical(self, message: str) -> None: ...
    def fatal(self, message: str) -> None: ...
    def failure(self, message: str) -> None: ...
    def exception(self, message: str) -> None: ...

class PrintLoggerFactory:
    """Factory for creating PrintLogger instances."""
    
    def __init__(self, file=None):
        """
        Initialize PrintLoggerFactory.
        
        Args:
            file (file-like, optional): Default file for created loggers
        """
    
    def __call__(self, *args) -> PrintLogger:
        """
        Create a PrintLogger instance.
        
        Args:
            *args: Arguments (ignored)
            
        Returns:
            PrintLogger: New PrintLogger instance
        """

Write Logger

Logger that writes messages to a file using direct write operations, offering better performance than PrintLogger.

class WriteLogger:
    """
    Logger that writes log messages directly to a file.
    
    Uses direct write() calls followed by flush() for better
    performance compared to PrintLogger.
    """
    
    def __init__(self, file=None):
        """
        Initialize WriteLogger.
        
        Args:
            file (file-like, optional): File object to write to.
                                       Defaults to sys.stdout if None.
        """
    
    def msg(self, message: str) -> None:
        """
        Write a log message and flush.
        
        Args:
            message (str): Message to write
        """
    
    # Alias methods for different log levels
    def log(self, message: str) -> None: ...
    def debug(self, message: str) -> None: ...
    def info(self, message: str) -> None: ...
    def warn(self, message: str) -> None: ...
    def warning(self, message: str) -> None: ...
    def err(self, message: str) -> None: ...
    def error(self, message: str) -> None: ...
    def critical(self, message: str) -> None: ...
    def fatal(self, message: str) -> None: ...
    def failure(self, message: str) -> None: ...
    def exception(self, message: str) -> None: ...

class WriteLoggerFactory:
    """Factory for creating WriteLogger instances."""
    
    def __init__(self, file=None):
        """
        Initialize WriteLoggerFactory.
        
        Args:
            file (file-like, optional): Default file for created loggers
        """
    
    def __call__(self, *args) -> WriteLogger:
        """
        Create a WriteLogger instance.
        
        Args:
            *args: Arguments (ignored)
            
        Returns:
            WriteLogger: New WriteLogger instance
        """

Bytes Logger

Logger that writes byte messages directly to a binary file or stream.

class BytesLogger:
    """
    Logger that writes log messages as bytes to a file.
    
    Designed for binary output or when you need to control
    the exact byte representation of log messages.
    """
    
    def __init__(self, file=None):
        """
        Initialize BytesLogger.
        
        Args:
            file (file-like, optional): Binary file object to write to.
                                       Defaults to sys.stdout.buffer if None.
        """
    
    def msg(self, message: bytes) -> None:
        """
        Write a log message as bytes.
        
        Args:
            message (bytes): Message bytes to write
        """
    
    # Alias methods for different log levels
    def log(self, message: bytes) -> None: ...
    def debug(self, message: bytes) -> None: ...
    def info(self, message: bytes) -> None: ...
    def warn(self, message: bytes) -> None: ...
    def warning(self, message: bytes) -> None: ...
    def err(self, message: bytes) -> None: ...
    def error(self, message: bytes) -> None: ...
    def critical(self, message: bytes) -> None: ...
    def fatal(self, message: bytes) -> None: ...
    def failure(self, message: bytes) -> None: ...
    def exception(self, message: bytes) -> None: ...

class BytesLoggerFactory:
    """Factory for creating BytesLogger instances."""
    
    def __init__(self, file=None):
        """
        Initialize BytesLoggerFactory.
        
        Args:
            file (file-like, optional): Default binary file for created loggers
        """
    
    def __call__(self, *args) -> BytesLogger:
        """
        Create a BytesLogger instance.
        
        Args:
            *args: Arguments (ignored)
            
        Returns:
            BytesLogger: New BytesLogger instance
        """

Usage Examples

Basic PrintLogger Usage

import structlog
from structlog import PrintLoggerFactory, processors
import sys

# Configure with PrintLogger
structlog.configure(
    processors=[
        processors.TimeStamper(fmt="iso"),
        processors.JSONRenderer()
    ],
    wrapper_class=structlog.BoundLogger,
    logger_factory=PrintLoggerFactory(),
)

logger = structlog.get_logger()
logger.info("Application started", version="1.0.0")
# Prints JSON to stdout

Writing to a File

import structlog
from structlog import WriteLoggerFactory, processors

# Open log file
with open("application.log", "w") as log_file:
    # Configure with file output
    structlog.configure(
        processors=[
            processors.TimeStamper(),
            processors.JSONRenderer()
        ],
        wrapper_class=structlog.BoundLogger,
        logger_factory=WriteLoggerFactory(file=log_file),
    )
    
    logger = structlog.get_logger()
    logger.info("Starting application", pid=1234)
    logger.info("Application ready", port=8080)
    # Messages written directly to file

High-Performance Logging

import structlog
from structlog import WriteLoggerFactory, processors
import sys

# Configure for high-performance output
structlog.configure(
    processors=[
        processors.JSONRenderer()  # Minimal processing
    ],
    wrapper_class=structlog.BoundLogger,
    logger_factory=WriteLoggerFactory(file=sys.stdout),
    cache_logger_on_first_use=True,  # Cache for performance
)

logger = structlog.get_logger()

# Fast logging in tight loops
for i in range(1000):
    logger.info("Processing item", item_id=i, batch="A")

Bytes Logger Usage

import structlog
from structlog import BytesLoggerFactory, processors
import sys

def bytes_json_renderer(logger, name, event_dict):
    """Custom processor that returns bytes."""
    import json
    json_str = json.dumps(event_dict)
    return json_str.encode('utf-8') + b'\n'

# Configure with BytesLogger
structlog.configure(
    processors=[
        processors.TimeStamper(),
        bytes_json_renderer
    ],
    wrapper_class=structlog.BoundLogger,
    logger_factory=BytesLoggerFactory(),
)

logger = structlog.get_logger()
logger.info("Binary logging", data="test")
# Outputs bytes to stdout.buffer

Multiple Output Destinations

import structlog
from structlog import WriteLoggerFactory, processors
import sys

class MultiFileLogger:
    """Custom logger that writes to multiple files."""
    
    def __init__(self, files):
        self.files = files
    
    def msg(self, message):
        for file in self.files:
            file.write(message + '\n')
            file.flush()
    
    # Add alias methods
    def info(self, message): return self.msg(message)
    def error(self, message): return self.msg(message)
    def warning(self, message): return self.msg(message)

class MultiFileLoggerFactory:
    def __init__(self, files):
        self.files = files
    
    def __call__(self, *args):
        return MultiFileLogger(self.files)

# Open multiple output files
with open("app.log", "w") as app_log, \
     open("audit.log", "w") as audit_log:
    
    structlog.configure(
        processors=[
            processors.TimeStamper(),
            processors.JSONRenderer()
        ],
        wrapper_class=structlog.BoundLogger,
        logger_factory=MultiFileLoggerFactory([app_log, audit_log, sys.stdout]),
    )
    
    logger = structlog.get_logger()
    logger.info("Message goes to all three destinations")

Error Stream Logging

import structlog
from structlog import WriteLoggerFactory, processors
import sys

# Configure to write errors to stderr
structlog.configure(
    processors=[
        processors.TimeStamper(),
        processors.add_log_level,
        processors.JSONRenderer()
    ],
    wrapper_class=structlog.BoundLogger,
    logger_factory=WriteLoggerFactory(file=sys.stderr),
)

logger = structlog.get_logger()
logger.error("Application error", error_code="E001", component="database")
# Error written to stderr

Log Rotation with Output Loggers

import structlog
from structlog import WriteLoggerFactory, processors
import os
from datetime import datetime

class RotatingFileLogger:
    """Logger that rotates files based on size or time."""
    
    def __init__(self, base_filename, max_bytes=1024*1024):
        self.base_filename = base_filename
        self.max_bytes = max_bytes
        self.current_file = None
        self.bytes_written = 0
        self._open_new_file()
    
    def _open_new_file(self):
        if self.current_file:
            self.current_file.close()
        
        timestamp = datetime.now().strftime("%Y%m%d_%H%M%S")
        filename = f"{self.base_filename}.{timestamp}"
        self.current_file = open(filename, "w")
        self.bytes_written = 0
    
    def msg(self, message):
        message_bytes = len(message.encode('utf-8'))
        
        if self.bytes_written + message_bytes > self.max_bytes:
            self._open_new_file()
        
        self.current_file.write(message + '\n')
        self.current_file.flush()
        self.bytes_written += message_bytes + 1
    
    def info(self, message): return self.msg(message)
    def error(self, message): return self.msg(message)

class RotatingLoggerFactory:
    def __init__(self, base_filename, max_bytes=1024*1024):
        self.logger = RotatingFileLogger(base_filename, max_bytes)
    
    def __call__(self, *args):
        return self.logger

# Configure with rotating logger
structlog.configure(
    processors=[
        processors.TimeStamper(),
        processors.JSONRenderer()
    ],
    wrapper_class=structlog.BoundLogger,
    logger_factory=RotatingLoggerFactory("app.log", max_bytes=1024),
)

logger = structlog.get_logger()
for i in range(100):
    logger.info("Log message", iteration=i, data="x" * 50)
# Creates multiple rotated log files

Buffered Output Logger

import structlog
from structlog import processors
import sys
import threading
import time

class BufferedLogger:
    """Logger that buffers messages and flushes periodically."""
    
    def __init__(self, file=None, buffer_size=100, flush_interval=5.0):
        self.file = file or sys.stdout
        self.buffer = []
        self.buffer_size = buffer_size
        self.flush_interval = flush_interval
        self.lock = threading.Lock()
        self.last_flush = time.time()
    
    def msg(self, message):
        with self.lock:
            self.buffer.append(message)
            
            # Flush if buffer is full or enough time has passed
            now = time.time()
            if (len(self.buffer) >= self.buffer_size or 
                now - self.last_flush >= self.flush_interval):
                self._flush()
    
    def _flush(self):
        if self.buffer:
            for message in self.buffer:
                self.file.write(message + '\n')
            self.file.flush()
            self.buffer.clear()
            self.last_flush = time.time()
    
    def info(self, message): return self.msg(message)
    def error(self, message): return self.msg(message)

class BufferedLoggerFactory:
    def __init__(self, **kwargs):
        self.logger = BufferedLogger(**kwargs)
    
    def __call__(self, *args):
        return self.logger

# Configure with buffered logger
structlog.configure(
    processors=[
        processors.TimeStamper(),
        processors.JSONRenderer()
    ],
    wrapper_class=structlog.BoundLogger,
    logger_factory=BufferedLoggerFactory(buffer_size=10, flush_interval=2.0),
)

logger = structlog.get_logger()
for i in range(5):
    logger.info("Buffered message", count=i)
    time.sleep(0.5)
# Messages are buffered and flushed together

Install with Tessl CLI