support stdlib %-style log records in LogFormatter

laser_prynter/log.py

@@ -23,19 +23,25 @@
 
     logger.debug('This is a debug message', 'arg1', 'arg2', {'key': 'value'})
     # {"timestamp": "2024-12-09T15:05:43.904749+10:00", "msg": "This is a debug message", "event": {"args": ["arg1", "arg2"], "key": "value"}}
+
+    # stdlib / third-party loggers (e.g. uvicorn) using %-style formatting
+    # are rendered automatically — no subclass needed:
+    #   logging.getLogger('uvicorn').info('Listening on %s:%d', '0.0.0.0', 8000)
+    #   # {"timestamp": "...", "msg": "Listening on 0.0.0.0:8000", "event": {}}
     ```
 '''
 
-from datetime import datetime
 import json
 import logging
-from logging.handlers import  TimedRotatingFileHandler
 import os
 import sys
-from typing import Any, TextIO
+from datetime import datetime
+from logging.handlers import TimedRotatingFileHandler
+from typing import TextIO
 
 from laser_prynter.pp import _json_default
 
+
 class LogLevel:
     'An enum type for log levels.'
     CRITICAL = logging.CRITICAL
@@ -50,48 +56,61 @@ class LogLevel:
 
 class LogFormatter(logging.Formatter):
     'Custom log formatter that formats log messages as JSON, aka "Structured Logging".'
-    def __init__(self, defaults: dict = {}):
+
+    def __init__(self, defaults: dict | None = None, access_fields: bool = True):
         '''
         Initializes the log formatter with optional default context.
         - `defaults` is a dictionary of default context values to include in every log message.
+        - `access_fields` promotes uvicorn access-log tuples to structured event fields.
         '''
-        self.defaults = defaults
+        self.defaults = defaults or {}
+        self.access_fields = access_fields
         super().__init__()
 
     def format(self, record: logging.LogRecord) -> str:
         'Formats the log message as JSON.'
-
-        args: tuple | list | None = None
-        kwargs: Any = {}
-
-        if isinstance(record.args, tuple):
-            if len(record.args) == 1:
-                args = record.args
-            elif len(record.args) > 1:
-                *args, kwargs = record.args
-        elif isinstance(record.args, dict):
-            kwargs = record.args
-
-        record.msg = json.dumps(
-            {
-                'timestamp': datetime.now().astimezone().isoformat(),
-                'level':     record.levelname,
-                'name':      record.name,
-                'msg':       record.msg,
-                'event':     {'args': args} if args else {} | kwargs or {},
-                **({'context': self.defaults} if self.defaults else {}),
-            },
-            default=_json_default,
-        )
+        message, event = self._render(record)
+        payload = {
+            'timestamp': datetime.now().astimezone().isoformat(),
+            'level':     record.levelname,
+            'name':      record.name,
+            'msg':       message,
+            'event':     event,
+            **({'context': self.defaults} if self.defaults else {}),
+        }
+        record.msg = json.dumps(payload, default=_json_default)
         record.args = ()
         return super().format(record)
 
+    def _render(self, record: logging.LogRecord) -> tuple[str, dict]:
+        args = record.args
+
+        if self.access_fields and record.name.endswith('access') \
+                and isinstance(args, tuple) and len(args) == 5:
+            client, method, path, http_version, status = args
+            return record.getMessage(), {
+                'client': client, 'method': method, 'path': path,
+                'http_version': http_version, 'status': status,
+            }
+
+        if isinstance(args, dict):
+            return record.msg, dict(args)
+
+        if isinstance(args, tuple) and args and isinstance(args[-1], dict):
+            *positional, context = args
+            event = dict(context) if isinstance(context, dict) else {}
+            if positional:
+                event['args'] = list(positional)
+            return record.msg, event
+
+        return record.getMessage(), {}
+
 
 def _getLogger(
     name:     str,
-    level:    int                   = logging.CRITICAL,
-    handlers: list[logging.Handler] = [],
-    context:  dict                  = {},
+    level:    int                              = logging.CRITICAL,
+    handlers: list[logging.Handler] | None     = None,
+    context:  dict | None                      = None,
 ) -> logging.Logger:
     '''
     Creates a logger with the given name, level, and handlers.
@@ -121,21 +140,21 @@ def _getLogger(
             logger.removeHandler(handler)
 
     # add the new handlers
-    for handler in handlers:
+    for handler in (handlers or []):
         logger.addHandler(handler)
 
     if logger.handlers:
         # only set the first handler to use the custom formatter
-        logger.handlers[0].setFormatter(LogFormatter(defaults=context))
+        logger.handlers[0].setFormatter(LogFormatter(defaults=context or {}))
 
     return logger
 
 def getLogger(
     name:     str,
-    level:    int                 = -1,
-    stream:   TextIO       = sys.stdout,
-    files:    dict[int, str] = {},
-    context:  dict                = {},
+    level:    int                       = -1,
+    stream:   TextIO                    = sys.stdout,
+    files:    dict[int, str] | None     = None,
+    context:  dict | None               = None,
 ) -> logging.Logger:
     '''
     Creates a logger with the given name, level, and handlers.
@@ -162,7 +181,7 @@ def getLogger(
         handler.setLevel(level)
         handlers.append(handler)
 
-    for flevel, filename in files.items():
+    for flevel, filename in (files or {}).items():
         fhandler = TimedRotatingFileHandler(
             filename, when='midnight', backupCount=7, encoding='utf-8',
         )

pyproject.toml

@@ -1,6 +1,6 @@
 [project]
 name = "laser-prynter"
-version = "0.12.0"
+version = "0.13.0"
 authors = [{ name = "tmck-code", email = "tmck01@gmail.com" }]
 description = "terminal/cli/python helpers for colour and pretty-printing"
 readme = "README.md"

test/pp/test_log_formatter_stdlib.py

@@ -0,0 +1,136 @@
+import json
+import logging
+from typing import Any
+
+from laser_prynter.log import LogFormatter
+
+
+def _make_record(
+    name: str = 'test', level: int = logging.INFO, msg: str = '', args: Any = None,
+) -> logging.LogRecord:
+    record = logging.LogRecord(
+        name=name, level=level, pathname='', lineno=0,
+        msg=msg, args=None, exc_info=None,
+    )
+    if args is not None:
+        record.args = args
+    return record
+
+
+def _format(record: logging.LogRecord, **kwargs: Any) -> dict[str, Any]:
+    formatter = LogFormatter(**kwargs)
+    output = formatter.format(record)
+    return json.loads(output)  # type: ignore[no-any-return]
+
+
+class TestLaserPrynterStyle:
+    """G2: existing laser-prynter call conventions produce identical output."""
+
+    def test_plain_message(self) -> None:
+        record = _make_record(msg='This is a basic info message')
+        result = _format(record)
+        assert result['msg'] == 'This is a basic info message'
+        assert result['event'] == {}
+
+    def test_single_dict_context(self) -> None:
+        record = _make_record(msg='This is an info message', args={'key': 'value'})
+        result = _format(record)
+        assert result['msg'] == 'This is an info message'
+        assert result['event'] == {'key': 'value'}
+
+    def test_positionals_with_trailing_dict(self) -> None:
+        record = _make_record(msg='This is a debug message', args=('arg1', 'arg2', {'key': 'value'}))
+        result = _format(record)
+        assert result['msg'] == 'This is a debug message'
+        assert result['event'] == {'args': ['arg1', 'arg2'], 'key': 'value'}
+
+
+class TestStdlibPercentStyle:
+    """G1: stdlib %-style records are fully rendered."""
+
+    def test_uvicorn_startup(self) -> None:
+        record = _make_record(
+            name='uvicorn.error',
+            msg='Uvicorn running on %s://%s:%d (Press CTRL+C to quit)',
+            args=('http', '0.0.0.0', 8000),
+        )
+        result = _format(record)
+        assert result['msg'] == 'Uvicorn running on http://0.0.0.0:8000 (Press CTRL+C to quit)'
+        assert result['event'] == {}
+        assert '%s' not in result['msg']
+        assert '%d' not in result['msg']
+
+    def test_uvicorn_access_structured(self) -> None:
+        record = _make_record(
+            name='uvicorn.access',
+            msg='%s - "%s %s HTTP/%s" %s',
+            args=('127.0.0.1:5000', 'GET', '/health', '1.1', 200),
+        )
+        result = _format(record)
+        assert result['event']['client'] == '127.0.0.1:5000'
+        assert result['event']['method'] == 'GET'
+        assert result['event']['path'] == '/health'
+        assert result['event']['http_version'] == '1.1'
+        assert result['event']['status'] == 200
+        assert '%s' not in result['msg']
+
+    def test_uvicorn_access_fields_disabled(self) -> None:
+        record = _make_record(
+            name='uvicorn.access',
+            msg='%s - "%s %s HTTP/%s" %s',
+            args=('127.0.0.1:5000', 'GET', '/health', '1.1', 200),
+        )
+        result = _format(record, access_fields=False)
+        assert result['event'] == {}
+        assert '127.0.0.1:5000' in result['msg']
+        assert '200' in result['msg']
+
+    def test_stdlib_string_format_args(self) -> None:
+        record = _make_record(msg='Connected to %s on port %d', args=('localhost', 5432))
+        result = _format(record)
+        assert result['msg'] == 'Connected to localhost on port 5432'
+        assert result['event'] == {}
+
+    def test_no_args(self) -> None:
+        record = _make_record(msg='Simple message', args=None)
+        result = _format(record)
+        assert result['msg'] == 'Simple message'
+        assert result['event'] == {}
+
+    def test_empty_tuple_args(self) -> None:
+        record = _make_record(msg='Simple message', args=())
+        result = _format(record)
+        assert result['msg'] == 'Simple message'
+        assert result['event'] == {}
+
+
+class TestEdgeCases:
+    """Documented caveats and edge cases."""
+
+    def test_lone_dict_arg_treated_as_context(self) -> None:
+        """Caveat: a stdlib caller whose final %-arg is a dict gets read as context."""
+        record = _make_record(msg='payload=%s', args=({'a': 1},))
+        result = _format(record)
+        assert result['msg'] == 'payload=%s'
+        assert result['event'] == {'a': 1}
+
+    def test_defaults_included_as_context(self) -> None:
+        record = _make_record(msg='hi')
+        result = _format(record, defaults={'service': 'web'})
+        assert result['context'] == {'service': 'web'}
+
+    def test_defaults_none(self) -> None:
+        record = _make_record(msg='hi')
+        result = _format(record, defaults=None)
+        assert 'context' not in result
+
+    def test_access_fields_non_access_logger(self) -> None:
+        """5-tuple on a non-access logger is treated as stdlib %-style."""
+        record = _make_record(
+            name='myapp.server',
+            msg='%s %s %s %s %s',
+            args=('a', 'b', 'c', 'd', 'e'),
+        )
+        result = _format(record)
+        assert result['msg'] == 'a b c d e'
+        assert result['event'] == {}