FastAPI中的中间件，用于为每个请求生成UUID并发送到日志

Question

我正在快速API中寻找中间件，用于为每个请求生成UUID并将其发送到日志。

# To see the logs, run this in python interpreter
# import with_logger
# with_logger.main()
import uuid
from logging import getLogger
import logging.config
from yaml import load, FullLoader
from fastapi import FastAPI
import uvicorn
app = FastAPI()

# @app.middleware("http")
@app.get("/")
def gen_uuid():
# For every get request generate uuid
global uuid_var
uuid_var = uuid.uuid4()
return uuid_var

log_dict = {
'version': 1,
'disable_existing_loggers': False,
'formatters': {
'standard': {
'format': '%(uuid_var)s %(asctime)s %(levelname)-8s %(name)-15s %(message)s -Thread id: %(thread)d  Thread name: %(threadName)s'
},
},
'handlers': {
'default': {
'level': 'INFO',
'formatter': 'standard',
'class': 'logging.StreamHandler',
},
'file_handler': {
'level': 'INFO',
'filename': 'mylogfile.log',
'class': 'logging.FileHandler',
'formatter': 'standard'
}
},
'loggers': {
'': {
'handlers': ['file_handler'],
'level': 'INFO',
'propagate': True
},
}
}

class ContextFilter(logging.Filter):
def filter(self, record):
record.uuid_var = gen_uuid()
return True

def main():
logging.config.dictConfig(log_dict)
handler = logging.StreamHandler()
handler.formatter = logging.Formatter(
"%(uuid_var)s %(asctime)s %(levelname)-8s %(name)-15s %(message)s -Thread id: %(thread)d  Thread name: %(threadName)s")
handler.addFilter(ContextFilter())
logger = getLogger(__name__)
logger.addHandler(handler)
logger.warning(" This is a warning message")
uvicorn.run(app, host="127.0.0.1", port=8080)

我需要将uuid发送到X-Request-ID头，我将使用python日志库从那里获取它，比如

@app.middleware("http")
async def get_x_request_id_header(*, x_request_id: str = Header(None)):
# return x-request-id header
return {"X-Request-ID": x_request_id}

我需要在中间件中完成所有这些。在Fast API中是否已经存在类似的中间件？

我需要在中间件中生成uuid，比如

@app.middleware("http")
def gen_uuid():
# For every get request generate uuid
global uuid_var
uuid_var = uuid.uuid4()
return uuid_var

但不能做到这一点。我收到错误，比如gen_uuid接受2个参数，但传递0个参数

Answer 1

我刚刚完成了一个中间件的实现。希望它仍然对你有用(或者对将来偶然发现的其他人有用(。

这是回购：https://github.com/snok/asgi-correlation-id

Answer 2

中间件需要两个参数：request:request，call_nextrequest是api请求，callnext是处理请求的实际调用类似这样的东西：

@app.middleware("http")
async def init_requestvars(request: fastapi.Request, call_next):
# Customize that SimpleNamespace with hatever you need
initial_g = types.SimpleNamespace()
requestvars.request_global.set(initial_g)
response = await call_next(request)
return response