FastAPI+Structlog日志格式统一：纯文本转JSON问题求助

API 探路者

2026-7-17

问题：FastAPI请求日志无法转为JSON格式，如何统一全日志JSON输出？

问题背景

用FastAPI和Structlog开发时，想把所有日志统一转成JSON格式，方便日志聚合平台读取处理。现在自己用Structlog打的日志已经是JSON，但FastAPI的请求访问日志还是纯文本，达不到全JSON输出的预期。

当前日志输出

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"}

预期日志输出

{
    "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:62155",
    "method": "GET",
    "endpoint": "/api/preface",
    "protocol": "HTTP/1.1",
    "status_code": 200,
    "status": "OK"
}
{
    "level": "INFO",
    "IP": "127.0.0.1:62155",
    "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"}

现有代码

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(_, __, 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,
                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 requests
from requests.auth import HTTPBasicAuth
from requests import Response

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="Push to NFS Success",
    )
    return {"Hello World !!"}

解决方案

核心问题是FastAPI依赖的Uvicorn日志属于Python标准logging模块输出，之前的配置只处理了Structlog自身的日志，没接管标准日志系统。需要做以下三步：

1. 修改Structlog配置，接管标准logging

更新configure_structlog方法，让标准日志也走Structlog的处理器链：

@staticmethod
def configure_structlog(logging_level, logger_name):
    # 初始化标准logging的基础配置
    logging.basicConfig(level=logging_level)
    
    structlog.configure(
        processors=[
            StructLogTest.logger_name,
            structlog.threadlocal.merge_threadlocal,
            structlog.processors.CallsiteParameterAdder(),
            # 新增：解析Uvicorn访问日志的处理器，放在前面
            StructLogTest.parse_uvicorn_access_log,
            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"),
            # 适配标准logging的格式要求，最后转JSON
            structlog.stdlib.ProcessorFormatter.wrap_for_formatter,
        ],
        wrapper_class=structlog.make_filtering_bound_logger(logging_level),
        context_class=dict,
        logger_factory=structlog.stdlib.LoggerFactory(),
    )

    # 给标准logging的处理器绑定Structlog的JSON格式化器
    formatter = structlog.stdlib.ProcessorFormatter(
        processor=structlog.processors.JSONRenderer(serializer=orjson.dumps),
        foreign_pre_chain=[
            structlog.processors.add_log_level,
            structlog.processors.TimeStamper(fmt="iso", utc=True, key="time-iso"),
        ],
    )

    for handler in logging.getLogger().handlers:
        handler.setFormatter(formatter)
    return structlog

2. 添加Uvicorn访问日志解析处理器

在StructLogTest类中新增一个方法，解析Uvicorn的纯文本日志，提取结构化字段：

@staticmethod
def parse_uvicorn_access_log(_, __, event_dict):
    message = event_dict.get("event")
    # 匹配Uvicorn默认访问日志的正则表达式
    import re
    pattern = r'INFO:\s+(\S+)\s+-\s+"(\w+) (\S+) (\S+)" (\d+) (\w+)'
    match = re.match(pattern, message)
    if match:
        # 提取并添加结构化字段
        event_dict["IP"] = match.group(1)
        event_dict["method"] = match.group(2)
        event_dict["endpoint"] = match.group(3)
        event_dict["protocol"] = match.group(4)
        event_dict["status_code"] = int(match.group(5))
        event_dict["status"] = match.group(6)
        # 删除原始的非结构化消息
        del event_dict["event"]
    return event_dict

3. 启动时禁用Uvicorn自带日志配置

在启动FastAPI的文件中，指定Uvicorn不使用自带的日志配置：

import uvicorn
from fastapi import FastAPI
from main import router

app = FastAPI()
app.include_router(router)

if __name__ == "__main__":
    uvicorn.run(
        app,
        host="0.0.0.0",
        port=8000,
        log_config=None,  # 禁用Uvicorn自带日志配置，用我们的Structlog配置
    )

关键说明

FastAPI的请求日志本质是Uvicorn的访问日志，属于Python标准logging模块的输出，之前的配置没覆盖这部分。
通过structlog.stdlib.LoggerFactory和ProcessorFormatter，可以让标准日志也经过Structlog处理转成JSON。
自定义的parse_uvicorn_access_log处理器负责把Uvicorn的纯文本日志拆分成IP、请求方法等结构化字段，最终输出符合预期的JSON格式。

内容的提问来源于stack exchange，提问作者Goku

FastAPI+Structlog日志格式统一：纯文本转JSON问题求助

问题背景 #

当前日志输出 #

预期日志输出 #

现有代码 #

struct.py #

main.py #

解决方案 #

1. 修改Structlog配置，接管标准logging #

2. 添加Uvicorn访问日志解析处理器 #

3. 启动时禁用Uvicorn自带日志配置 #

关键说明 #