README.md

Two Python classes that rotate log files and compress them

compressed_log_handler.py

import os
import gzip
import lzma
import shutil
from logging.handlers import RotatingFileHandler, TimedRotatingFileHandler


class CompressedRotatorMixin:
    """
    Mixin class that adds compression functionality to rotating file handlers.
    
    Uses the namer and rotator extension points to add compression support.
    """
    
    def _setup_compression(self, compression):
        """Set up compression by configuring namer and rotator."""
        if compression not in ('gzip', 'xz'):
            raise ValueError("compression must be 'gzip' or 'xz'")
        
        self.compression = compression
        ext = '.gz' if compression == 'gzip' else '.xz'
        
        # Set up namer to add compression extension
        def namer(default_name):
            return default_name + ext
        
        self.namer = namer
        self.rotator = self._compress_rotator
    
    def _compress_rotator(self, source, dest):
        """Compress the rotated log file."""
        if self.compression == 'gzip':
            with open(source, 'rb') as f_in:
                with gzip.open(dest, 'wb') as f_out:
                    shutil.copyfileobj(f_in, f_out)
        elif self.compression == 'xz':
            with open(source, 'rb') as f_in:
                with lzma.open(dest, 'wb') as f_out:
                    shutil.copyfileobj(f_in, f_out)
        
        os.remove(source)


class CompressedRotatingFileHandler(CompressedRotatorMixin, RotatingFileHandler):
    """
    A RotatingFileHandler that compresses rotated log files.
    
    Args:
        filename: Path to the log file
        compression: Compression format - 'gzip' or 'xz' (default: 'gzip')
    `filename` and all other arguments are passed to RotatingFileHandler
    """
    
    def __init__(self, filename, /, **kwargs):
        compression = kwargs.pop('compression', 'gzip')
        super().__init__(filename, **kwargs)
        self._setup_compression(compression)


class CompressedTimedRotatingFileHandler(CompressedRotatorMixin, TimedRotatingFileHandler):
    """
    A TimedRotatingFileHandler that compresses rotated log files.

    Args:
        filename: Path to the log file
        compression: Compression format - 'gzip' or 'xz' (default: 'gzip')
    `filename` and all other arguments are passed to TimedRotatingFileHandler
    """

    def __init__(self, filename, /, **kwargs):
        compression = kwargs.pop('compression', 'gzip')
        super().__init__(filename, **kwargs)
        self._setup_compression(compression)


# Example usage
if __name__ == '__main__':
    import logging
    
    # Example 1: Size-based rotation with gzip compression
    logger1 = logging.getLogger('size_based')
    logger1.setLevel(logging.DEBUG)
    
    handler1 = CompressedRotatingFileHandler(
        'app.log',
        maxBytes=4*1024,  # 4KB
        backupCount=5,
        compression='gzip'
    )
    
    formatter = logging.Formatter('%(asctime)s - %(name)s - %(levelname)s - %(message)s')
    handler1.setFormatter(formatter)
    logger1.addHandler(handler1)
    
    # Example 2: Time-based rotation with xz compression
    logger2 = logging.getLogger('time_based')
    logger2.setLevel(logging.DEBUG)
    
    handler2 = CompressedTimedRotatingFileHandler(
        'app_timed.log',
        when='midnight',  # Rotate at midnight
        interval=1,
        backupCount=7,  # Keep 7 days
        compression='xz'
    )
    
    handler2.setFormatter(formatter)
    logger2.addHandler(handler2)
    
    # Test logging
    for i in range(1000):
        logger1.info(f'Size-based log message {i}')
        logger2.info(f'Time-based log message {i}')
    
    print("Logging complete.")
    print("Size-based: Check for app.log and compressed backups (app.log.1.gz, etc.)")
    print("Time-based: Check for app_timed.log and compressed backups (app_timed.log.YYYY-MM-DD.xz, etc.)")