返回

FastAPI日志:自定义JSON格式化与字段调整

python

FastAPI 日志结构自定义:JSON格式化与字段调整

当使用FastAPI开发应用程序时,日志记录是关键。默认的日志格式可能无法满足所有需求,比如需要将日志输出为特定结构的JSON格式,或者需要包含不同的字段。本文将探讨如何调整FastAPI应用程序和访问日志的结构和字段。

重新格式化应用程序日志

FastAPI 使用 Uvicorn 作为默认的 ASGI 服务器,而 Uvicorn 又使用 Python 标准库的 logging 模块进行日志输出。要自定义应用日志的格式,核心在于配置 logging 模块。目标是改变输出格式,将其包装在 XYZ 键之下,并包含自定义字段。

方案:自定义日志处理器和格式化器

以下步骤可达成目标:

  1. 定义日志格式化器: 创建一个继承自 logging.Formatter 的类,并重写 format 方法。 在此方法中,将原始日志记录数据转换为需要的 JSON 结构。
  2. 创建日志处理器: 将格式化器绑定到一个自定义处理器。此处理器可以是输出到 stdout,也可以是其他目的地,比如文件或者第三方服务。
  3. 配置Logger: 获取fastapi 的logger,删除所有默认的handler,添加自定义的处理器,并设置级别。
  4. 设置Uvicorn logger: 取消禁用 uvicorn.access,保证可以打印access日志,配置logger.

代码示例:

import logging
import json
import sys
from datetime import datetime
from fastapi import FastAPI
from uvicorn.config import LOGGING_CONFIG


class CustomJSONFormatter(logging.Formatter):
    def format(self, record):
        log_data = {
            "XYZ": {
                "log": {
                    "level": record.levelname.lower(),
                    "type": "app",
                    "timestamp": datetime.fromtimestamp(record.created).isoformat(),
                    "file": record.pathname,
                    "line": record.lineno,
                    "threadId": record.thread,
                    "message": record.getMessage(),
                }
            }
        }
        return json.dumps(log_data)

app = FastAPI()
# 禁用 Uvicorn 默认的日志记录
LOGGING_CONFIG["loggers"]["uvicorn.access"]["propagate"] = True

# 获取根logger,移除默认的Handler,配置自定义logger
logger = logging.getLogger("uvicorn")
logger.handlers.clear()

custom_handler = logging.StreamHandler(sys.stdout)
custom_handler.setFormatter(CustomJSONFormatter())
logger.addHandler(custom_handler)
logger.setLevel(logging.INFO)


# 其他配置和路由定义...
@app.get("/items/{item_id}")
async def read_item(item_id: int):
   logger.info(f"Item {item_id} requested")
   return {"item_id": item_id}


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

操作步骤:

  1. 复制上述代码到一个python 文件中(如 main.py)
  2. 运行 python main.py
  3. 访问/items/1, 然后查看终端输出。

重新格式化访问日志

对于访问日志的格式化,FastAPI 使用了中间件,可以通过自定义中间件修改访问日志格式。需要扩展访问日志并使用自定义的结构和字段。

方案:自定义访问日志中间件

可以通过创建一个中间件来完成这项工作。拦截请求,记录需要的详细信息。然后将日志以定制的 JSON 格式输出。

代码示例:

import logging
import json
import sys
import time
from datetime import datetime
from fastapi import FastAPI, Request, Response
from uvicorn.config import LOGGING_CONFIG

class CustomJSONFormatter(logging.Formatter):
    def format(self, record):
        log_data = {
            "XYZ": {
                "log": {
                    "level": record.levelname.lower(),
                    "type": "app",
                    "timestamp": datetime.fromtimestamp(record.created).isoformat(),
                    "file": record.pathname,
                    "line": record.lineno,
                    "threadId": record.thread,
                    "message": record.getMessage(),
                }
            }
        }
        return json.dumps(log_data)

app = FastAPI()


class CustomAccessJSONFormatter(logging.Formatter):
   def __init__(self):
        super().__init__()
        self.access_log_logger = logging.getLogger("uvicorn.access")

   def format(self, record):

      if not hasattr(record, 'scope') or not hasattr(record, 'response'):
        return None #Skip
      scope = record.scope
      response = record.response

      log_data = {
          "XYZ": {
              "log": {
                  "level": record.levelname.lower(),
                  "type": "access",
                  "timestamp": datetime.fromtimestamp(record.created).isoformat(),
                  "message": f"{scope['method']} {scope['path']} {response.status_code} {int((record.end_time-record.start_time)*1000)}ms",
              },
              "req": {
                  "url": scope.get('path'),
                  "headers": dict(scope.get('headers', [])),
                  "method": scope.get('method'),
                  "httpVersion": scope.get('http_version'),
                   "originalUrl": scope.get("raw_path"),
                  "query": dict(scope.get("query_string")),
              },
              "res":{
                   "statusCode": response.status_code,
                  "body": response.body,

              }
          }
      }
      return json.dumps(log_data)

@app.middleware("http")
async def custom_middleware(request: Request, call_next):
   start_time = time.time()
   response = await call_next(request)
   end_time = time.time()
   setattr(response,"body",None)
   body = b""
   async for chunk in response.body_iterator:
        body += chunk
   setattr(response, "body",json.loads(body.decode('utf-8')) if len(body)>0 and "application/json" in response.headers.get('content-type', '') else body.decode('utf-8') if len(body)>0 else None )
   setattr(response, 'status_code', response.status_code ) #save original value
   record = logging.makeLogRecord({'scope': request.scope,
                          'response':response,
                           'start_time': start_time,
                          'end_time':end_time
                          })
   formatter=CustomAccessJSONFormatter()
   formatted_message=formatter.format(record)
   if formatted_message is not None:
      access_logger=logging.getLogger("uvicorn.access")
      access_logger.info(formatted_message)
   return response



LOGGING_CONFIG["loggers"]["uvicorn.access"]["propagate"] = False  #  不能设置为 True ,避免重复记录 access 日志,应该添加自定义 handler,在custom_middleware中记录
logger = logging.getLogger("uvicorn")
logger.handlers.clear()
custom_handler = logging.StreamHandler(sys.stdout)
custom_handler.setFormatter(CustomJSONFormatter())
logger.addHandler(custom_handler)
logger.setLevel(logging.INFO)


@app.get("/app/health")
async def health_check():
    return {"status":"OK", "statusCode": 200}


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

操作步骤:

  1. 将上面的代码保存到python 文件 (如 main.py)
  2. 运行 python main.py
  3. 访问/app/health,然后在终端查看输出。

其他安全建议:

  • 日志敏感信息: 不要在日志中包含敏感数据,如密码,API 密钥,或者用户的私有信息。考虑使用加密,数据混淆,或者过滤中间件,避免记录此类数据。
  • 日志大小: 日志过多可能会影响性能。要确保配置合理的日志轮转和存储策略。
  • 安全审计: 定期审计日志记录配置。确保没有记录意料之外的数据,没有过多的安全漏洞暴露在日志中。

以上展示了自定义FastAPI应用和访问日志格式与字段的一些基本方案。这些示例为处理不同的日志记录需求提供了坚实的基础,你可以根据具体的项目需求进行调整和扩展。