Skip to content

Unify Python logging for a Gunicorn/Uvicorn/FastAPI application

I recently started playing with FastAPI and HTTPX, and I am deploying my app with Gunicorn and Uvicorn workers.

But when serving, the logs from each component looks quite different from the others. I want them to all look the same, so I can easily read them or exploit them in something like Kibana.

After a lot of hours trying to understand how Python logging works, and how to override libraries' logging settings, here is what I have...

A single run.py file! I didn't want to split logging configuration, Gunicorn configuration, and the rest of the code into multiple files, as it was harder to wrap my head around it.

Gunicorn + Uvicorn version¤

Everything is contained in this single file:

import os
import logging
import sys

from gunicorn.app.base import BaseApplication
from gunicorn.glogging import Logger
from loguru import logger

from my_app.app import app


LOG_LEVEL = logging.getLevelName(os.environ.get("LOG_LEVEL", "DEBUG"))
JSON_LOGS = True if os.environ.get("JSON_LOGS", "0") == "1" else False
WORKERS = int(os.environ.get("GUNICORN_WORKERS", "5"))


class InterceptHandler(logging.Handler):
    def emit(self, record):
        # Get corresponding Loguru level if it exists
        try:
            level = logger.level(record.levelname).name
        except ValueError:
            level = record.levelno

        # Find caller from where originated the logged message
        frame, depth = logging.currentframe(), 2
        while frame.f_code.co_filename == logging.__file__:
            frame = frame.f_back
            depth += 1

        logger.opt(depth=depth, exception=record.exc_info).log(level, record.getMessage())


class StubbedGunicornLogger(Logger):
    def setup(self, cfg):
        handler = logging.NullHandler()
        self.error_logger = logging.getLogger("gunicorn.error")
        self.error_logger.addHandler(handler)
        self.access_logger = logging.getLogger("gunicorn.access")
        self.access_logger.addHandler(handler)
        self.error_logger.setLevel(LOG_LEVEL)
        self.access_logger.setLevel(LOG_LEVEL)


class StandaloneApplication(BaseApplication):
    """Our Gunicorn application."""

    def __init__(self, app, options=None):
        self.options = options or {}
        self.application = app
        super().__init__()

    def load_config(self):
        config = {
            key: value for key, value in self.options.items()
            if key in self.cfg.settings and value is not None
        }
        for key, value in config.items():
            self.cfg.set(key.lower(), value)

    def load(self):
        return self.application


if __name__ == '__main__':
    intercept_handler = InterceptHandler()
    # logging.basicConfig(handlers=[intercept_handler], level=LOG_LEVEL)
    # logging.root.handlers = [intercept_handler]
    logging.root.setLevel(LOG_LEVEL)

    seen = set()
    for name in [
        *logging.root.manager.loggerDict.keys(),
        "gunicorn",
        "gunicorn.access",
        "gunicorn.error",
        "uvicorn",
        "uvicorn.access",
        "uvicorn.error",
    ]:
        if name not in seen:
            seen.add(name.split(".")[0])
            logging.getLogger(name).handlers = [intercept_handler]

    logger.configure(handlers=[{"sink": sys.stdout, "serialize": JSON_LOGS}])

    options = {
        "bind": "0.0.0.0",
        "workers": WORKERS,
        "accesslog": "-",
        "errorlog": "-",
        "worker_class": "uvicorn.workers.UvicornWorker",
        "logger_class": StubbedGunicornLogger
    }

    StandaloneApplication(app, options).run()

If you are in a hurry, copy-paste it, change the Gunicorn options at the end, and try it!

If you're not, I will explain each part below.


import os
import logging
import sys

from gunicorn.app.base import BaseApplication
from gunicorn.glogging import Logger
from loguru import logger

This part is easy, we simply import the things we need. The Gunicorn BaseApplication so we can run Gunicorn directly from this script, and its Logger that we will override a bit. We are using Loguru later in the code, to have a pretty log format, or to serialize them.


from my_app.app import app

In my project, I have a my_app package with an app module. My FastAPI application is declared in this module, something like app = FastAPI().


LOG_LEVEL = logging.getLevelName(os.environ.get("LOG_LEVEL", "DEBUG"))
JSON_LOGS = True if os.environ.get("JSON_LOGS", "0") == "1" else False
WORKERS = int(os.environ.get("GUNICORN_WORKERS", "5"))

We setup some values from environment variables, useful for development vs. production setups. JSON_LOGS tells if we should serialize the logs to JSON, and WORKERS tells how many workers we want to have.


class InterceptHandler(logging.Handler):
    def emit(self, record):
        # Get corresponding Loguru level if it exists
        try:
            level = logger.level(record.levelname).name
        except ValueError:
            level = record.levelno

        # Find caller from where originated the logged message
        frame, depth = logging.currentframe(), 2
        while frame.f_code.co_filename == logging.__file__:
            frame = frame.f_back
            depth += 1

        logger.opt(depth=depth, exception=record.exc_info).log(level, record.getMessage())

This code is copy-pasted from Loguru's documentation! This handler will be used to intercept the logs emitted by libraries and re-emit them through Loguru.


class StubbedGunicornLogger(Logger):
    def setup(self, cfg):
        handler = logging.NullHandler()
        self.error_logger = logging.getLogger("gunicorn.error")
        self.error_logger.addHandler(handler)
        self.access_logger = logging.getLogger("gunicorn.access")
        self.access_logger.addHandler(handler)
        self.error_logger.setLevel(LOG_LEVEL)
        self.access_logger.setLevel(LOG_LEVEL)

This code was copied from this GitHub comment by @dcosson. Thanks! It will allow us to override Gunicorn's own logging configuration so its logs can be formatted like the rest.


class StandaloneApplication(BaseApplication):
    """Our Gunicorn application."""

    def __init__(self, app, options=None):
        self.options = options or {}
        self.application = app
        super().__init__()

    def load_config(self):
        config = {
            key: value for key, value in self.options.items()
            if key in self.cfg.settings and value is not None
        }
        for key, value in config.items():
            self.cfg.set(key.lower(), value)

    def load(self):
        return self.application

This code is taken from Gunicorn's documentation. We declare a simple Gunicorn application that we will be able to run. It accepts all Gunicorn's options.


if __name__ == '__main__':
    intercept_handler = InterceptHandler()
    # logging.basicConfig(handlers=[intercept_handler], level=LOG_LEVEL)
    # logging.root.handlers = [intercept_handler]
    logging.root.setLevel(LOG_LEVEL)

We simply instantiate our interception handler, and set the log level on the root logger.

Once again, I fail to understand how this works exactly, as the two commented lines have no impact on the result. I did a lot of trial and error and ended up with something working, but I cannot entirely explain why. The idea here was to set the handler on the root logger so it intercepts everything, but it was not enough (logs were not all intercepted).


    seen = set()
    for name in [
        *logging.root.manager.loggerDict.keys(),
        "gunicorn",
        "gunicorn.access",
        "gunicorn.error",
        "uvicorn",
        "uvicorn.access",
        "uvicorn.error",
    ]:
        if name not in seen:
            seen.add(name.split(".")[0])
            logging.getLogger(name).handlers = [intercept_handler]

Here we iterate on all the possible loggers declared by libraries to override their handlers with our interception handler. This is where we actually configure every logger to behave the same.

For a reason that I fail to understand, Gunicorn and Uvicorn do not appear in the root logger manager, so we have to hardcode them in the list.

We also use a set to avoid setting the interception handler on the parent of a logger that is already configured, because otherwise logs would be emitted twice or more. I'm not sure this code can handle levels of nested loggers deeper than two.


    logger.configure(handlers=[{"sink": sys.stdout, "serialize": JSON_LOGS}])

Here we configure Loguru to write on the standard output, and to serialize logs if needed.

At some point I was also using activation=[("", True)] (see Loguru's docs), but it seems it's not required either.


    options = {
        "bind": "0.0.0.0",
        "workers": WORKERS,
        "accesslog": "-",
        "errorlog": "-",
        "worker_class": "uvicorn.workers.UvicornWorker",
        "logger_class": StubbedGunicornLogger
    }

    StandaloneApplication(app, options).run()

Finally, we set our Gunicorn options, wiring things up, and run our application!


Well, I'm not really proud of this code, but it works!

logs

logs_json

Uvicorn-only version¤

Added Nov 11, 2020.

The Uvicorn-only version is way more simple. Note that since this post was published the first time, a new Uvicorn version was released, which contained a fix for its logging configuration: could be in 0.11.6 (Don't override the root logger) or 0.12.0 (Dont set log level for root logger).

This simplifies a lot the setup_logging function, which now makes more sense and is easier to understand:

import os
import logging
import sys

from uvicorn import Config, Server
from loguru import logger

LOG_LEVEL = logging.getLevelName(os.environ.get("LOG_LEVEL", "DEBUG"))
JSON_LOGS = True if os.environ.get("JSON_LOGS", "0") == "1" else False


class InterceptHandler(logging.Handler):
    def emit(self, record):
        # Get corresponding Loguru level if it exists
        try:
            level = logger.level(record.levelname).name
        except ValueError:
            level = record.levelno

        # Find caller from where originated the logged message
        frame, depth = logging.currentframe(), 2
        while frame.f_code.co_filename == logging.__file__:
            frame = frame.f_back
            depth += 1

        logger.opt(depth=depth, exception=record.exc_info).log(level, record.getMessage())


def setup_logging():
    # intercept everything at the root logger
    logging.root.handlers = [InterceptHandler()]
    logging.root.setLevel(LOG_LEVEL)

    # remove every other logger's handlers
    # and propagate to root logger
    for name in logging.root.manager.loggerDict.keys():
        logging.getLogger(name).handlers = []
        logging.getLogger(name).propagate = True

    # configure loguru
    logger.configure(handlers=[{"sink": sys.stdout, "serialize": JSON_LOGS}])


if __name__ == '__main__':
    server = Server(
        Config(
            "my_app.app:app",
            host="0.0.0.0",
            log_level=LOG_LEVEL,
        ),
    )

    # setup logging last, to make sure no library overwrites it
    # (they shouldn't, but it happens)
    setup_logging()

    server.run()