如何获得不同结构和不同字段的JSON格式的FastAPI应用程序的控制台日志？
问题描述
我有一个fast API应用程序，我想在其中使用以下数据以JSON格式将默认日志写入STDOUT：
应用程序日志应如下所示：
{
"XYZ": {
"log": {
"level": "info",
"type": "app",
"timestamp": "2022-01-16T08:30:08.181Z",
"file": "api/predictor/predict.py",
"line": 34,
"threadId": 435454,
"message": "API Server started on port 8080 (development)"
}
}
}
访问日志应如下所示：
{
"XYZ": {
"log": {
"level": "info",
"type": "access",
"timestamp": "2022-01-16T08:30:08.181Z",
"message": "GET /app/health 200 6ms"
},
"req": {
"url": "/app/health",
"headers": {
"host": "localhost:8080",
"user-agent": "curl/7.68.0",
"accept": "*/*"
},
"method": "GET",
"httpVersion": "1.1",
"originalUrl": "/app/health",
"query": {}
},
"res": {
"statusCode": 200,
"body": {
"statusCode": 200,
"status": "OK"
}
}
}
}
我尝试的内容
我尝试使用json-logging包来实现这一点。使用this示例，我能够访问json中的请求日志并更改结构。但我找不到如何访问和更改应用程序日志。
当前输出日志结构
{"written_at": "2022-01-28T09:31:38.686Z", "written_ts": 1643362298686910000, "msg":
"Started server process [12919]", "type": "log", "logger": "uvicorn.error", "thread":
"MainThread", "level": "INFO", "module": "server", "line_no": 82, "correlation_id":
"-"}
{"written_at": "2022-01-28T09:31:38.739Z", "written_ts": 1643362298739838000, "msg":
"Started server process [12919]", "type": "log", "logger": "uvicorn.error", "thread":
"MainThread", "level": "INFO", "module": "server", "line_no": 82, "correlation_id":
"-"}
{"written_at": "2022-01-28T09:31:38.739Z", "written_ts": 1643362298739951000, "msg":
"Waiting for application startup.", "type": "log", "logger": "uvicorn.error",
"thread": "MainThread", "level": "INFO", "module": "on", "line_no": 45,
"correlation_id": "-"}
解决方案
您可以通过使用内置记录器模块创建自定义Formatter来完成此操作。您可以在记录消息时使用extra参数来传递上下文信息，如URL和标头。Python's JSON module已经在dump()函数中实现了漂亮的打印JSON数据，使用indent参数定义缩进级别。下面是一个使用自定义格式化程序以您在问题中描述的格式记录消息的工作示例。例如，对于App日志，使用logger.info('sample log message')，而对于访问日志，使用logger.info('sample log message', extra={'extra_info': get_extra_info(request)})。通过将request实例传递给get_extra_info()方法，您可以提取上面提到的信息。有关更多LogRecord属性，请查看here。下面的示例也使用FileHandler将消息记录在日志文件中。如果您不需要它，可以在get_logger()方法中将其注释掉。
app_logger.py
import logging, sys
def get_file_handler(formatter, log_filename):
file_handler = logging.FileHandler(log_filename)
file_handler.setLevel(logging.DEBUG)
file_handler.setFormatter(formatter)
return file_handler
def get_stream_handler(formatter):
stream_handler = logging.StreamHandler(sys.stdout)
stream_handler.setLevel(logging.DEBUG)
stream_handler.setFormatter(formatter)
return stream_handler
def get_logger(name, formatter, log_filename = "logfile.log"):
logger = logging.getLogger(name)
logger.setLevel(logging.DEBUG)
logger.addHandler(get_file_handler(formatter, log_filename))
logger.addHandler(get_stream_handler(formatter))
return logger
app_logger_formatter.py
import json, logging
def get_app_log(record):
json_obj = {'XYZ': {'log': {
'level': record.levelname,
'type': 'app',
'timestamp': record.asctime,
#'filename': record.filename,
'pathname': record.pathname,
'line': record.lineno,
'threadId': record.thread,
'message': record.message
}}}
return json_obj
def get_access_log(record):
json_obj = {'XYZ': {'log': {
'level': record.levelname,
'type': 'access',
'timestamp': record.asctime,
'message': record.message},
'req': record.extra_info['req'],
'res': record.extra_info['res']}}
return json_obj
class CustomFormatter(logging.Formatter):
def __init__(self, formatter):
logging.Formatter.__init__(self, formatter)
def format(self, record):
logging.Formatter.format(self, record)
if not hasattr(record, 'extra_info'):
return json.dumps(get_app_log(record), indent=2)
else:
return json.dumps(get_access_log(record), indent=2)
app.py
import app_logger
from app_logger_formatter import CustomFormatter
from fastapi import FastAPI, Request
import uvicorn
app = FastAPI()
formatter = CustomFormatter('%(asctime)s')
logger = app_logger.get_logger(__name__, formatter)
def get_extra_info(request):
return {'req': {
'url': request.url.path,
'headers': {'host': request.headers['host'],
'user-agent': request.headers['user-agent'],
'accept': request.headers['accept']},
'method': request.method,
'httpVersion': request.scope['http_version'],
'originalUrl': request.url.path,
'query': {}
},
'res': {'statusCode': 200, 'body': {'statusCode': 200,
'status': 'OK'}}}
@app.get("/foo")
def foo(request: Request):
logger.info('GET ' + request.url.path, extra={'extra_info': get_extra_info(request)})
return "success"
if __name__ == '__main__':
logger.info("Server started listening on port: 8000")
uvicorn.run(app, host='127.0.0.1', port=8000)
更新%1
记录请求/响应的更好方法是添加FastAPI Middleware，它接受到达应用程序的每个请求，以及返回之前的每个响应。示例如下(app_logger.py和app_logger_Formatter.py保持不变)。
app.py
import app_logger
from app_logger_formatter import CustomFormatter
from fastapi import FastAPI, Request, Response
import uvicorn
from http import HTTPStatus
app = FastAPI()
formatter = CustomFormatter('%(asctime)s')
logger = app_logger.get_logger(__name__, formatter)
status_reasons = {x.value:x.name for x in list(HTTPStatus)}
def get_extra_info(request: Request, response: Response):
return {'req': {
'url': request.url.path,
'headers': {'host': request.headers['host'],
'user-agent': request.headers['user-agent'],
'accept': request.headers['accept']},
'method': request.method,
'httpVersion': request.scope['http_version'],
'originalUrl': request.url.path,
'query': {}
},
'res': {'statusCode': response.status_code, 'body': {'statusCode': response.status_code,
'status': status_reasons.get(response.status_code)}}}
@app.middleware("http")
async def log_request(request: Request, call_next):
response = await call_next(request)
logger.info(request.method + ' ' + request.url.path, extra={'extra_info': get_extra_info(request, response)})
return response
@app.get("/foo")
def foo(request: Request):
return "success"
if __name__ == '__main__':
logger.info("Server started listening on port: 8000")
uvicorn.run(app, host='127.0.0.1', port=8000)
更新%2
除了前面的更新之外，您还可以使用background task来记录数据，而不是让响应等待记录发生后再返回。后台任务将仅在"；发送后运行(根据Starlette documentation)。您可以定义一个任务函数在后台运行，用于写入日志数据，如下图所示：
def write_log_data(request, response):
logger.info(request.method + ' ' + request.url.path, extra={'extra_info': get_extra_info(request, response)})
然后，在中间件内部，您可以将(上面)任务函数传递给background tasks，如下所示(信用给this post)：
from starlette.background import BackgroundTask
@app.middleware("http")
async def log_request(request: Request, call_next):
response = await call_next(request)
response.background = BackgroundTask(write_log_data, request, response)
return response
输出：
{
"XYZ": {
"log": {
"level": "INFO",
"type": "app",
"timestamp": "2022-01-28 10:46:01,904",
"pathname": ".../app.py",
"line": 33,
"threadId": 408,
"message": "Server started listening on port: 8000"
}
}
}
{
"XYZ": {
"log": {
"level": "INFO",
"type": "access",
"timestamp": "2022-01-28 10:46:03,587",
"message": "GET /foo"
},
"req": {
"url": "/foo",
"headers": {
"host": "127.0.0.1:8000",
"user-agent": "Mozilla/5.0 ...",
"accept": "text/html,..."
},
"method": "GET",
"httpVersion": "1.1",
"originalUrl": "/foo",
"query": {}
},
"res": {
"statusCode": 200,
"body": {
"statusCode": 200,
"status": "OK"
}
}
}
}
相关文章