Fluentlog

Opinionated structured logging for Python with a fluent API.

• API inspired by zerolog
• JSON output format
• OpenTelemetry naming conventions when relevant
• Near zero-cost for disabled log levels

Installation

pip install fluentlog

Getting Started

Simple example

import fluentlog

log = fluentlog.Logger().bind().int("request_id", 1).logger()

log.info().str("user", "jmcs").int("uid", 42).msg("user logged in")
# {"level":"INFO","request_id":1,"user":"jmcs","uid":42,"message":"user logged in"}

# Disabled levels have near-zero overhead
log.debug().func(expensive_func).msg("debug info")  # expensive_func is never called

Log Levels

fluentlog supports the following log levels, from more to less critical:

• FATAL: Errors the application can't recover from
• ERROR: Errors that make the current context fail, but not the entire application
• WARNING: Recoverable errors
• INFO: Expected lifecycle events and relevant business signals
• DEBUG: Internal details useful while diagnosing behavior during development
• TRACE: Very fine-grained execution details, usually only useful for deep debugging

You can set the log level for your logger either in the constructor or using a fluent method:

import fluentlog

log = fluentlog.Logger(level=fluentlog.Level.DEBUG)

# or

log = fluentlog.Logger().set_level(fluentlog.Level.DEBUG)

You can also set the log level using a string, this is useful for reading the log level from an environment variable or a configuration file:

import os

import fluentlog

value_from_env = os.environ.get("LOG_LEVEL", "INFO")
log = fluentlog.Logger().set_level(fluentlog.Level.from_str(value_from_env))

Field Types

Immutable types (bool, bytes, float, int, path, date/datetime, timedelta, str)

log.info().bool("is_valid", True).int("user_id", 42).str("username", "jmcs").path(
    "file_path", "/path/to/file"
).time("timestamp", datetime.now()).timedelta("duration", timedelta(seconds=30)).msg(
    "User info"
)

Adds a field of the corresponding type to the context. Since these types are immutable, they are referenced directly without copying, which is more performant.

Dict and List

log.info().dict("user", {"id": 42, "name": "jmcs"}).list(
    "roles", ["admin", "user"]
).msg("Logging a dictionary and a list")

Adds a dictionary and a list field to the context. The dictionary and list are deep-copied to prevent mutations after the fact from affecting the log output, so it has a negative impact on performance.

Exception

try:
    1 / 0
except ZeroDivisionError as e:
    log.error().exception(e).msg("An error occurred")

Stores the exception details in the event fields:

• exception.type: the type of the exception (e.g. ValueError)
• exception.message: the message of the exception
• exception.stacktrace: the stack trace of the exception

Any

obj = SomeComplexObject()
log.info().any("object", obj).msg("Logging a complex object")

Adds a field with any value to the context. If the value is not JSON serializable, it will be converted to a string using repr() when the event is finalized.

Since this method accepts any value, it will be deep-copied to prevent mutations after the fact from affecting the log output, which negatively impacts performance.

Hooks

Hooks are functions that are called with the event context before the event is finalized, allowing for custom processing and enrichment of the event.

Arbitrary/Custom hook

def add_user_info(event: fluentlog.Event) -> None:
    user = get_current_user()  # example, potentially expensive operation
    event.str("user", user.name).int("user_id", user.id)


log.info().func(add_user_info).msg("Logging with a custom hook")

Runs the function if the log level is enabled. The function receives the event as an argument and can add fields to it.

Caller info

log.info().caller().msg("Logging with caller info")

Identifies the caller of the log method and adds it to the log fields, with the following fields:

• code.file.path: The full path of the file containing the caller.
• code.function.name: The name of the function containing the caller.
• code.line.number: The line number of the caller in the source code.

The optional skip parameter can be used to skip additional stack frames if the caller is wrapped in helper functions.

Timestamp

log.info().timestamp().msg("Logging with a timestamp")

Adds a timestamp field to the event with the current time in ISO 8601 format. For loggers, this is processed at output time. For events, this is processed when the timestamp() method is called.

Add multiple fields from a dictionary

log = log.bind().fields({"context": "example"}).logger()
fields = {"user": "jmcs", "uid": 42, "context": "example"}
log.info().fields(fields).msg("Logging with multiple fields from a dictionary")

Adds multiple fields to the event from a dictionary. The dictionary is deep-copied to prevent mutations after the event is sent from affecting the logged data.

Global Logger

Fluentlog supports a global logger that can be configured and used across modules without needing to pass it around.

import fluentlog


def main():
    log = (
        fluentlog.Logger()
        .set_level(fluentlog.Level.DEBUG)
        .bind()
        .str("context", "example")
        .logger()
    )
    fluentlog.set_global_logger(log)
    some_func()


def some_func():
    log = fluentlog.get_global_logger()
    log.debug().msg("From func")
    # {"level":"DEBUG", "context": "example", "message": "From func"}
    log.info().msg("From func")
    # {"level":"INFO", "context": "example", "message": "From func"}

Logging context

Fluentlog supports context-based logger passing, which allows for preserving logging context across function boundaries without having to pass the logger explicitly.

import fluentlog


def some_func():
    log = fluentlog.context()
    log.info().msg("From func")


def main():
    log = fluentlog.context().bind().str("context", "example").logger()
    some_func()
    # {"level":"INFO", "message": "From func"}
    with fluentlog.context_logger(log):
        some_func()
        # {"level":"INFO", "context": "example", "message": "From func"}


main()

Standard Logging Handler

Fluentlog provides a standard logging handler that allows using Fluentlog as a drop-in replacement for the standard logging module.

import logging
import fluentlog


def setup_logging():
    log = (
        fluentlog.Logger()
        .set_level(fluentlog.Level.DEBUG)
        .bind()
        .str("context", "example")
        .logger()
    )
    fluentlog.set_global_logger(log)

    # Create a standard logging handler that uses Fluentlog
    handler = fluentlog.FluentlogHandler(log)
    # CAVEAT: there are two independent level gates to configure: the standard
    # logging level (below) AND the fluentlog Logger's own level (set above via
    # set_level). Both must be at least as permissive as the lowest level you
    # want to capture, otherwise messages may be silently dropped by whichever
    # gate is stricter.
    logging.basicConfig(level=logging.DEBUG, handlers=[handler])

    logging.info("This is a standard logging message")
    # {"level":"INFO", "context": "example", "message": "This is a standard logging message"}

This handler recognizes the standard log levels (DEBUG, INFO, WARNING/WARN, ERROR, CRITICAL/FATAL). Any other level name is mapped to INFO.

Testing

Fluentlog provides testing utilities to allow testing logging behaviour easily.

from my_module import my_function

import pytest
from fluentlog.testing import TestingLogger


def test_my_function():
    log = TestingLogger()
    with fluentlog.context_logger(log):
        # my_function() is expected to log 42 events, including a specific message and bind an extra field to the log context
        my_function()

    assert log.count_events() == 42
    assert log.has_message("Very specific message")
    assert log.count_events_with_message("Not so specific message") == 10
    assert log.has_event(
        {"level": "INFO", "extra": 123, "message": "Function executed"}
    )
    event = log.get_event_by_message("Function executed")
    assert event is not None
    assert event["extra"] == 123
    assert len(log.get_events()) == 42
    assert log.get_field("extra") == 123

    # the log can be reset to clear all events
    log.reset()
    assert log.count_events() == 0

Instead of resetting the log manually, you can also use TestingLogger as a context manager, which will automatically reset the log when exiting the context:

def test_my_function():
    log = TestingLogger()
    with log:
        with fluentlog.context_logger(log):
            my_function()
        assert log.count_events() == 42
    assert log.count_events() == 0

If you use the global logger you can also replace it temporarily during a test:

def test_my_function():
    log = TestingLogger()
    with with_global_logger(log):
        # global logger is `log`
        my_function()
    # global logger is back to the previous logger

Performance

Benchmarks show ~2-3x faster than stdlib logging with formatted output, with greater advantages when log levels are filtered.

Design decisions

Why use different methods for different types?

Using different methods for different types allows for optimising serialization strategies for mutable and immutable types. For example, dict() and list() deep-copy their arguments to prevent mutations after the event is logged from affecting the output, while int() and str() can safely reference immutable values directly without copying.

Why dummy events for disabled log levels?

Having dummy events achieves near-zero overhead, as we can avoid unnecessary processing without having to check the log level everywhere.

Why OpenTelemetry naming conventions?

I use OpenTelemetry for distributed tracing, and like consistent and precise naming, even when it comes at the cost of verbosity.

Why no formatted messages?

Formatted messages are familiar because that's how traditional logging usually works. But for structured logs they are a trap, as important data gets buried in strings instead of proper fields, which makes filtering and querying harder.

Why context-based logger passing?

Preserving logging context across boundaries is essential in complex applications, but having bound context inside a function is useful too. Context-based logger passing allows for both options and keeps things purposeful while avoiding cluttering application APIs.