Uvicorn override current log configuration to JSON.

[Chel2000]

I'm using structlog for logging application logs that enables easier loading of logs to the centralized logging service, Splunk as well.
The uvicorn/fastAPI logging is using standard text-based logging.
What is the best way to override that to use my structlog configuration and print all logs in JSON format?

Example:

Current Output

INFO:     127.0.0.1:62154 - "GET /api/preface HTTP/1.1" 200 OK
INFO:     127.0.0.1:62154 - "GET /loader.json HTTP/1.1" 200 OK
INFO:     127.0.0.1:62155 - "GET /hello_world HTTP/1.1" 200 OK
{"key":"test_key","message":"Push to NFS Success","event":"Testing Fast API..","logger":"test_my_api","filename":"main.py","func_name":"Hello_World","process":23760,"module":"docker","thread":23140,"pathname":"D:\\my_work\\fast_api\\main.py","process_name":"SpawnProcess-1","level":"info","time-iso":"2023-06-30T15:25:03.113400Z"}

Expected Output

{ "level" : "INFO" , IP : 127.0.0.1:62154 , method : GET, endpoint: /api/preface, protocol : HTTP/1.1, status_code : 200, status : OK }
{ "level" : "INFO" , IP : 127.0.0.1:62154 , method : GET, endpoint: /loader.json, protocol : HTTP/1.1, status_code : 200, status : OK }
{ "level" : "INFO" , IP : 127.0.0.1:62154 , method : GET, endpoint: /hello_world, protocol : HTTP/1.1, status_code : 200, status : OK }

{"key":"test_key","message":"Push to NFS Success","event":"Testing Fast API..","logger":"test_my_api","filename":"main.py","func_name":"Hello_World","process":23760,"module":"docker","thread":23140,"pathname":"D:\\my_work\\fast_api\\main.py","process_name":"SpawnProcess-1","level":"info","time-iso":"2023-06-30T15:25:03.113400Z"}

I'm sharing the piece of code if that may prove useful.

struct.py

import orjson
import structlog
import logging

## Added only the necessary context.
class StructLogTest:
    def __init__(self, logging_level=logging.DEBUG, logger_name="test"):
        self.logging_level = logging_level
        self.logger_name = logger_name
        StructLogTest.logger_name_var = self.logger_name
        self._configure_structlog(self.logging_level, self.logger_name)

    def logger_name_def(_, __, event_dict):
        event_dict["test_log"] = StructLogTest.logger_name_var
        return event_dict


    @staticmethod
    def _configure_structlog(logging_level, logger_name):
        structlog.configure(
            processors=[
                StructLogTest.logger_name_def,
                structlog.threadlocal.merge_threadlocal,
                structlog.processors.CallsiteParameterAdder(),
                structlog.processors.add_log_level,
                structlog.stdlib.PositionalArgumentsFormatter(),
                structlog.processors.StackInfoRenderer(),
                structlog.processors.format_exc_info,
                structlog.processors.TimeStamper(fmt="iso", utc=True, key="time-iso"),
                structlog.processors.JSONRenderer(serializer=orjson.dumps),
            ],
            wrapper_class=structlog.make_filtering_bound_logger(logging_level),
            context_class=dict,
            logger_factory=structlog.BytesLoggerFactory(),
        )
        return structlog

    def define_Logger(self, *args, **kwargs):
        return structlog.get_logger(*args, **kwargs)

    def info(self, message, *args, **kwargs):
        return structlog.get_logger().info(message, *args, **kwargs)
    
    and other methods so on..    

main.py

from struct import StructLogTest
from fastapi import APIRouter
import uvicorn

log = StructLogTest(logger_name="test_my_api")
log = log.get_Logger()

@router.get("/hello_world")
def Hello_World():
    logg = log.bind(key=test_key)
    logg.info(
        "Testing Fast API..",
        message=some_other_meaningful_function.dump(),
    )
    return {" Hello World !! "}

[Kludex]

Does this help: https://gist.github.com/Kludex/8e6d30de151f1d756c3c7364811c9429 ?

If it doesn't, I'll check with more attention what you are asking.

[Chel2000]

Ah okay, thanks.

[vladyslav-burylov]

@Kludex @Chel2000 - actually there is a way to redirect everything:

Below uses loguru + ecs_logging, however it should be possible to use structlog as well

import logging

import ecs_logging
import loguru
import uvicorn

from fastapi import FastAPI

# Configure your favourite logger
loguru.logger.remove()
loguru_handler = logging.StreamHandler()
loguru_handler.setFormatter(ecs_logging.StdlibFormatter())
loguru.logger.add(loguru_handler, format="{message}")


# Create a handler class to redirect messages from uvicorn to your favourite logger
class UvicornLoggingHandler(logging.Handler):
    def emit(self, record: logging.LogRecord) -> None:
        # Redirect messages to your favourite logger
        loguru.logger.opt(exception=record.exc_info).log(record.levelname, self.format(record).rstrip())


# redirects messages from uvicorn and sends them to MyLoggingHandler
__UVICORN_LOG_CONFIG = {
    "version": 1,
    "disable_existing_loggers": False,
    "handlers": {
        "uvicorn": {
            "class": f"{UvicornLoggingHandler.__module__}.{UvicornLoggingHandler.__qualname__}",
        },
    },
    "root": {
        "handlers": ["uvicorn"], "propagate": False, "level": "TRACE"
    }
}


app = FastAPI()


@app.get("/test")
async def test():
    loguru.logger.info("test")

uvicorn.run(app, host="127.0.0.1", port=8010, log_config=__UVICORN_LOG_CONFIG)

Output:

{"@timestamp":"2023-07-12T23:33:24.086Z","log.level":"info","message":"Started server process [50297]","ecs":{"version":"1.6.0"},"log":{"logger":"__main__","origin":{"file":{"line":20,"name":"the_bug_server.py"},"function":"emit"},"original":"Started server process [50297]"},"process":{"name":"MainProcess","pid":50297,"thread":{"id":7998184256,"name":"MainThread"}}}
{"@timestamp":"2023-07-12T23:33:24.090Z","log.level":"info","message":"Waiting for application startup.","ecs":{"version":"1.6.0"},"log":{"logger":"__main__","origin":{"file":{"line":20,"name":"the_bug_server.py"},"function":"emit"},"original":"Waiting for application startup."},"process":{"name":"MainProcess","pid":50297,"thread":{"id":7998184256,"name":"MainThread"}}}
{"@timestamp":"2023-07-12T23:33:24.090Z","log.level":"info","message":"Application startup complete.","ecs":{"version":"1.6.0"},"log":{"logger":"__main__","origin":{"file":{"line":20,"name":"the_bug_server.py"},"function":"emit"},"original":"Application startup complete."},"process":{"name":"MainProcess","pid":50297,"thread":{"id":7998184256,"name":"MainThread"}}}
{"@timestamp":"2023-07-12T23:33:24.096Z","log.level":"info","message":"Uvicorn running on http://127.0.0.1:8010 (Press CTRL+C to quit)","ecs":{"version":"1.6.0"},"log":{"logger":"__main__","origin":{"file":{"line":20,"name":"the_bug_server.py"},"function":"emit"},"original":"Uvicorn running on http://127.0.0.1:8010 (Press CTRL+C to quit)"},"process":{"name":"MainProcess","pid":50297,"thread":{"id":7998184256,"name":"MainThread"}}}
{"@timestamp":"2023-07-12T23:33:55.525Z","log.level":"info","message":"test","ecs":{"version":"1.6.0"},"log":{"logger":"__main__","origin":{"file":{"line":43,"name":"the_bug_server.py"},"function":"test"},"original":"test"},"process":{"name":"MainProcess","pid":50297,"thread":{"id":7998184256,"name":"MainThread"}}}
{"@timestamp":"2023-07-12T23:33:55.526Z","log.level":"info","message":"127.0.0.1:61320 - \"GET /test HTTP/1.1\" 200","ecs":{"version":"1.6.0"},"log":{"logger":"__main__","origin":{"file":{"line":20,"name":"the_bug_server.py"},"function":"emit"},"original":"127.0.0.1:61320 - \"GET /test HTTP/1.1\" 200"},"process":{"name":"MainProcess","pid":50297,"thread":{"id":7998184256,"name":"MainThread"}}}
{"@timestamp":"2023-07-12T23:34:07.960Z","log.level":"info","message":"127.0.0.1:61327 - \"GET /not_found HTTP/1.1\" 404","ecs":{"version":"1.6.0"},"log":{"logger":"__main__","origin":{"file":{"line":20,"name":"the_bug_server.py"},"function":"emit"},"original":"127.0.0.1:61327 - \"GET /not_found HTTP/1.1\" 404"},"process":{"name":"MainProcess","pid":50297,"thread":{"id":7998184256,"name":"MainThread"}}}

[paketb0te]

@vladyslav-burylov cool workaround, although it looks like the access logs are just wrapped as messages in this case (e.g. "message":"127.0.0.1:61320 - \"GET /test HTTP/1.1\").

Would be awesome if there was a way to log the "components" of these access logs directly, e.g. like {"client":"127.0.0.1:61320", "method":"get", "full_path":"/test", "http_version":"1.1", "status_code":"200"}, but my understanding of python's and uvicorn's logging system is... still inadequate, it seems :P

[paketb0te]

I got it somewhat working by subclassing jsonlogger.JsonFormatter from the python-json-logger package:

# foo.py

import logging
from copy import copy

import uvicorn
from fastapi import FastAPI
from pythonjsonlogger import jsonlogger


class UvicornJSONAccessFormatter(jsonlogger.JsonFormatter):
    def format(self, record: logging.LogRecord) -> str:
        recordcopy = copy(record)
        client_addr, method, full_path, http_version, status_code = recordcopy.args  # type: ignore[misc]
        recordcopy.__dict__.update(
            {
                "client_addr": client_addr,
                "method": method,
                "full_path": full_path,
                "http_version": http_version,
                "status_code": status_code,
            }
        )
        return super().format(record=recordcopy)


class UvicornJSONDefaultFormatter(jsonlogger.JsonFormatter):
    def format(self, record: logging.LogRecord) -> str:
        recordcopy = copy(record)
        recordcopy.__dict__.pop("color_message", None)
        return super().format(record=recordcopy)


LOGGING_CONFIG = {
    "version": 1,
    "disable_existing_loggers": False,
    "formatters": {
        "default": {
            "()": UvicornJSONDefaultFormatter,
        },
        "access": {
            "()": UvicornJSONAccessFormatter,
        },
    },
    "handlers": {
        "default": {
            "formatter": "default",
            "class": "logging.StreamHandler",
            "stream": "ext://sys.stderr",
        },
        "access": {
            "formatter": "access",
            "class": "logging.StreamHandler",
            "stream": "ext://sys.stdout",
        },
    },
    "loggers": {
        "uvicorn": {"handlers": ["default"], "level": "INFO", "propagate": False},
        "uvicorn.error": {"level": "INFO"},
        "uvicorn.access": {"handlers": ["access"], "level": "INFO", "propagate": False},
    },
}


app = FastAPI()


@app.get("/")
async def test():
    return "hello world"


if __name__ == "__main__":
    uvicorn.run(app, host="0.0.0.0", port=8000, log_config=LOGGING_CONFIG)

When running, this creates logs in the following format:

$ python foo.py 
{"message": "Started server process [15910]", "taskName": "Task-1"}
{"message": "Waiting for application startup.", "taskName": "Task-1"}
{"message": "Application startup complete.", "taskName": "Task-1"}
{"message": "Uvicorn running on http://0.0.0.0:8000 (Press CTRL+C to quit)", "taskName": "Task-1"}
{"message": "127.0.0.1:34296 - \"GET / HTTP/1.1\" 200", "taskName": "Task-3", "client_addr": "127.0.0.1:34296", "method": "GET", "full_path": "/", "http_version": "1.1", "status_code": 200}
^C{"message": "Shutting down", "taskName": "Task-1"}
{"message": "Waiting for application shutdown.", "taskName": "Task-1"}
{"message": "Application shutdown complete.", "taskName": "Task-1"}
{"message": "Finished server process [15910]", "taskName": "Task-1"}

[dolfinus]

Proposal of adding extra fields to log records, so they can be used with json formatter without any patching: #2154