Sink function to receive raw record, or formatter to return formatted record?

[NiklasRosenstein]

Hi! I'm looking to yield a very customized JSON format for each log record. I've been trying to use a custom sink or format function, but there are some issue with it:

• When using the sink function, you get only the formatted log message as a string or the (not entirely raw) log record as a JSON-string (with serialized=True)
• When using a format function, you get the log record but my issue is that the message is already formatted to replace parameters in the text.

I'd like to be able to receive the un-formatted log message and arguments.

I've found a bit of a dumb workaround:

@dataclass(kw_only=True, unsafe_hash=True)
class JsonFormatter:
    stream: TextIOWrapper = field(default_factory=lambda: sys.stderr)
    include_unsafe: bool = True
    indent: int | None = 2
    flush_on_write: bool = False

    def __call__(self, record: "_Record") -> None:
        safe_parameters = {}
        unsafe_parameters = {}
        for key, value in record["extra"].items():
            if isinstance(value, safe):
                safe_parameters[key] = value.value
            elif isinstance(value, unsafe):
                unsafe_parameters[key] = value.value
            else:
                unsafe_parameters[key] = value

        payload: JsonLogRecord = {
            "time": datetime.now(timezone.utc).isoformat(),
            "level": record["level"].name,
            "message": record["message"],
            "metadata": {
                "filename": record["file"].name,
                "pathname": record["file"].path,
                "funcname": record["function"],
                "levelno": record["level"].no,
                "lineno": record["line"],
                "process": record["process"].id,
                "processname": record["process"].name,
                "thread": record["thread"].id,
                "threadname": record["thread"].name,
            },
            "parameters": {
                "safe": safe_parameters,
                "unsafe": unsafe_parameters if self.include_unsafe else {},
            }
        }

        self.stream.write(json.dumps(payload, indent=self.indent, default=repr))
        self.stream.write("\n")
        if self.flush_on_write:
            self.stream.flush()
        return ""

This almost works, but the arguments in the message are already formatted at this point.

logger.remove()
logger.add(sys.stderr, format=JsonFormatter())
logger.info(
    "Hello, World! My name is {name}",
    name=safe("John Smith"),
    password="JaneSmith1234",
)

• Is there any way to prevent the message from being already formatted? Something like a "message formatter" function I can pass (and make it a no-op)?
• Can a formatter just output the fully formatted text instead of the log format to use with str.format()?

[mrkmcnamee]

Hi Niklas,

keyword arguments should not be used for formatting as this may be deprecated in future.
#1005 (comment)

Cheers,
Kevin

[Delgan]

Hi @NiklasRosenstein.

Is there any way to prevent the message from being already formatted? Something like a "message formatter" function I can pass (and make it a no-op)?

If you mean the message such as "Hello, World! My name is {name}" (ignoring the formatting of name), then this is not possible. The message formatting precedes everything else, and although I thought about making this configurable in the past, this would be technically challenging and not worth it in my opinion.
Besides, as @mrkmcnamee pointed out, keyword-arguments formatting will likely become deprecated.

Can a formatter just output the fully formatted text instead of the log format to use with str.format()?

No, but you can achieve something similar by assigning your formatted string to an extra attribute and returning it as the format:

from loguru import logger
from datetime import datetime, timezone
import sys
import json

def custom_json_serializer(record):
    custom_json = {
            "time": datetime.now(timezone.utc).isoformat(),
            "level": record["level"].name,
            "message": record["message"],
    }
    record["extra"]["custom_json"] = json.dumps(custom_json)
    return "{extra[custom_json]}\n"

logger.add(sys.stderr, format=custom_json_serializer)

logger.info("?")

The overhead of str.format() should be negligible.