Concurrent Log Handler async bg logging example

run_clh_test.py

import os

import asyncio
from contextlib import asynccontextmanager
import logging
import logging.config
from concurrent_log_handler.queue import setup_logging_queues, stop_queue_listeners

logger = logging.getLogger()


def my_logging_setup(log_file_name, use_async=False):
    logging_config = {
        "version": 1,
        "disable_existing_loggers": False,
        "formatters": {
            "default": {"format": "%(asctime)s %(levelname)s %(name)s %(message)s"},
            "formatter2": {
                "format": "[%(asctime)s][%(levelname)s][%(filename)s:%(lineno)s]"
                "[%(process)d][%(message)s]",
            },
        },
        # Set up our concurrent logger handler. Need one of these per unique file.
        "handlers": {
            "ch_log": {
                "level": "DEBUG",
                "class": "logging.StreamHandler",
                "formatter": "formatter2",
            },
            "fh_log": {
                "level": "DEBUG",
                "class": "concurrent_log_handler.ConcurrentRotatingFileHandler",
                # Example of a custom format for this log.
                "formatter": "formatter2",
                # 'formatter': 'default',
                "filename": log_file_name,
                # Optional: set an owner and group for the log file
                # 'owner': ['greenfrog', 'admin'],
                # Sets permissions to owner and group read+write
                "chmod": 0o0660,
                # Note: this is abnormally small to make it easier to demonstrate rollover.
                # A more reasonable value might be 10 MiB or 10485760
                "maxBytes": 5 * 1024 * 1024,
                # Number of rollover files to keep
                "backupCount": 10,
                # 'use_gzip': True,
            },
        },
        # Tell root logger to use our concurrent handler
        "root": {
            "handlers": ["ch_log", "fh_log"],
            "level": "DEBUG",
        },
    }

    logging.config.dictConfig(logging_config)

    if use_async:
        setup_logging_queues()

    return


@asynccontextmanager
async def lifespan(app):
    try:
        # logger = logging.getLogger()
        logger.setLevel(logging.INFO)

        # set log file name
        log_file_name = os.path.dirname(__file__) + "/test.log"
        log_file_name = os.path.expandvars(log_file_name)

        my_logging_setup(log_file_name, True)

        logger.info("Logger started...")

        yield

    except (KeyboardInterrupt, SystemExit) as e:
        # scheduler.shutdown(wait=False)
        print(f"Got: {e}")
        raise
    finally:
        stop_queue_listeners()

        logger.info("Logger ended")

        print("Server Stopped")


async def main_async_runner():
    print("Async runner started.")
    async with lifespan(None):
        print("Inside lifespan's 'yield' block (simulating app running).")

        # Simulate some logging activity
        logger.debug("This is a debug message from the app.")
        logger.info("This is an info message from the app.")
        logger.warning("This is a warning from the app.")

        # Try to log something similar to the problematic message
        # The bug report had "Arguments: None" associated with the error context.
        # Standard logging calls logger.info(message_string, *args)
        # If you log logger.info(message_string) -> record.args will be an empty tuple ()
        # If you log logger.info(message_string, None) -> record.args will be (None,)
        # It's rare for record.args to *be* None itself.
        logger.info('192.168.88.162:54558 - "GET /api/version HTTP/1.1" 200')

        # Let's try logging with an explicit None argument to see if it has an effect
        # This would make record.args = (None,)
        logger.error("An error message with a None argument: %s", None)

        # Simulate some work
        print("App: doing some work...")
        await asyncio.sleep(1)  # Simulate async work
        print("App: work done.")

        # You could add more complex logging or operations here to try and trigger the bug
        for i in range(3):
            logger.info(f"Logging loop iteration {i}")
            await asyncio.sleep(0.2)

    print("Async runner finished.")


if __name__ == "__main__":
    print("Starting script execution...")
    asyncio.run(main_async_runner())
    print("Script execution complete.")