返回
FastAPI日志:自定义JSON格式化与字段调整
python
2025-01-16 14:38:59
FastAPI 日志结构自定义:JSON格式化与字段调整
当使用FastAPI开发应用程序时,日志记录是关键。默认的日志格式可能无法满足所有需求,比如需要将日志输出为特定结构的JSON格式,或者需要包含不同的字段。本文将探讨如何调整FastAPI应用程序和访问日志的结构和字段。
重新格式化应用程序日志
FastAPI 使用 Uvicorn 作为默认的 ASGI 服务器,而 Uvicorn 又使用 Python 标准库的 logging
模块进行日志输出。要自定义应用日志的格式,核心在于配置 logging
模块。目标是改变输出格式,将其包装在 XYZ
键之下,并包含自定义字段。
方案:自定义日志处理器和格式化器
以下步骤可达成目标:
- 定义日志格式化器: 创建一个继承自
logging.Formatter
的类,并重写format
方法。 在此方法中,将原始日志记录数据转换为需要的 JSON 结构。 - 创建日志处理器: 将格式化器绑定到一个自定义处理器。此处理器可以是输出到
stdout
,也可以是其他目的地,比如文件或者第三方服务。 - 配置Logger: 获取fastapi 的logger,删除所有默认的handler,添加自定义的处理器,并设置级别。
- 设置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)
操作步骤:
- 复制上述代码到一个python 文件中(如 main.py)
- 运行
python main.py
- 访问
/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)
操作步骤:
- 将上面的代码保存到python 文件 (如 main.py)
- 运行
python main.py
- 访问
/app/health
,然后在终端查看输出。
其他安全建议:
- 日志敏感信息: 不要在日志中包含敏感数据,如密码,API 密钥,或者用户的私有信息。考虑使用加密,数据混淆,或者过滤中间件,避免记录此类数据。
- 日志大小: 日志过多可能会影响性能。要确保配置合理的日志轮转和存储策略。
- 安全审计: 定期审计日志记录配置。确保没有记录意料之外的数据,没有过多的安全漏洞暴露在日志中。
以上展示了自定义FastAPI应用和访问日志格式与字段的一些基本方案。这些示例为处理不同的日志记录需求提供了坚实的基础,你可以根据具体的项目需求进行调整和扩展。