在 Python 快速 API 中记录每个 API 请求的 UUID
Logging UUID per API request in Python FastAPI
我有一个纯粹的 python 程序包(我们称它为主要程序包),其中包含一些用于管理基础设施的功能。此外,我还创建了一个 FastAPI 服务,可以根据需要调用主模块来调用功能。
对于日志记录,我使用 loguru。 API 在启动时创建一个 loguru 实例,应用设置并设置通用 UUID(即 [main])。在对 API 的每个传入请求中,pre_request 函数都会生成一个新的 UUID 并调用 loguru 以使用该 UUID 进行配置。在请求结束时,UUID 被设置回默认 UUID [main].
我面临的问题是并发请求,新的 UUID 接管并且所有日志现在都使用最新配置的 UUID 写入。有没有一种方法可以在每个请求上实例化 loguru 模块,并确保并行处理的 API 请求不会发生交叉日志记录?
实施:
在主包的init.py中:
from loguru import logger
logger.remove() #to delete all existing default loggers
logger.add(filename, format, level, retention, rotation) #format
logger.configure(extra={"uuid": "main"})
在所有模块中,记录器被导入为
from loguru import logger
在 api/ 包中 - 在每个新请求中,我都有下面的代码块:
uuid = get_uuid() #calling util func to get a new uuid
logger.configure(uuid=uuid)
# Here onwards, all log messages contain this uuid
# At the end of the request, I configure it back to default uuid (i.e. "main")
configure 方法正在更新根记录器,我尝试使用 bind 方法代替,根据 loguru 文档,它可用于将额外的记录属性上下文化,但它似乎没有任何效果(我仍然查看默认 UUID,即“main”,仅当我使用 .configure 时 UUID 才会设置)。
关于我应该如何设置 UUID 的任何想法,以便对 API 的所有并发请求都有自己的 UUID?由于有多个子模块被调用来为一个 API 请求提供服务,并且所有子模块都有一些登录信息,因此我需要 UUID 为每个请求的所有模块保留。似乎每个 API 请求我都需要一个记录器实例,但我不确定如何正确实例化它以使其工作。
如果 API 正在处理一个请求,则当前的实现有效,但在处理超过 1 个调用时记录中断(因为记录的 UUID 是配置的最后一个)
我创建了这个中间件,它在路由调用之前使用上下文管理器用户的 UUID 配置记录器实例:
from loguru import logger
from contextvars import ContextVar
from starlette.middleware.base import BaseHTTPMiddleware
_request_id = ContextVar("request_id", default=None)
def get_request_id():
return _request_id.get()
class ContextualizeRequest(BaseHTTPMiddleware):
async def dispatch(self, request, call_next):
uuid = get_uuid()
request_id = _request_id.set(uuid) # set uuid to context variable
with logger.contextualize(uuid=get_request_id()):
try:
response = await call_next(request)
except Exception:
logger.error("Request failed")
finally:
_request_id.reset()
return response
唯一需要注意的是,如果您使用多线程,multi-processing,或创建新的事件循环(在内部调用中),该块中的记录器实例将没有此 UUID。
我有一个纯粹的 python 程序包(我们称它为主要程序包),其中包含一些用于管理基础设施的功能。此外,我还创建了一个 FastAPI 服务,可以根据需要调用主模块来调用功能。
对于日志记录,我使用 loguru。 API 在启动时创建一个 loguru 实例,应用设置并设置通用 UUID(即 [main])。在对 API 的每个传入请求中,pre_request 函数都会生成一个新的 UUID 并调用 loguru 以使用该 UUID 进行配置。在请求结束时,UUID 被设置回默认 UUID [main].
我面临的问题是并发请求,新的 UUID 接管并且所有日志现在都使用最新配置的 UUID 写入。有没有一种方法可以在每个请求上实例化 loguru 模块,并确保并行处理的 API 请求不会发生交叉日志记录?
实施:
在主包的init.py中:
from loguru import logger
logger.remove() #to delete all existing default loggers
logger.add(filename, format, level, retention, rotation) #format
logger.configure(extra={"uuid": "main"})
在所有模块中,记录器被导入为
from loguru import logger
在 api/ 包中 - 在每个新请求中,我都有下面的代码块:
uuid = get_uuid() #calling util func to get a new uuid
logger.configure(uuid=uuid)
# Here onwards, all log messages contain this uuid
# At the end of the request, I configure it back to default uuid (i.e. "main")
configure 方法正在更新根记录器,我尝试使用 bind 方法代替,根据 loguru 文档,它可用于将额外的记录属性上下文化,但它似乎没有任何效果(我仍然查看默认 UUID,即“main”,仅当我使用 .configure 时 UUID 才会设置)。
关于我应该如何设置 UUID 的任何想法,以便对 API 的所有并发请求都有自己的 UUID?由于有多个子模块被调用来为一个 API 请求提供服务,并且所有子模块都有一些登录信息,因此我需要 UUID 为每个请求的所有模块保留。似乎每个 API 请求我都需要一个记录器实例,但我不确定如何正确实例化它以使其工作。
如果 API 正在处理一个请求,则当前的实现有效,但在处理超过 1 个调用时记录中断(因为记录的 UUID 是配置的最后一个)
我创建了这个中间件,它在路由调用之前使用上下文管理器用户的 UUID 配置记录器实例:
from loguru import logger
from contextvars import ContextVar
from starlette.middleware.base import BaseHTTPMiddleware
_request_id = ContextVar("request_id", default=None)
def get_request_id():
return _request_id.get()
class ContextualizeRequest(BaseHTTPMiddleware):
async def dispatch(self, request, call_next):
uuid = get_uuid()
request_id = _request_id.set(uuid) # set uuid to context variable
with logger.contextualize(uuid=get_request_id()):
try:
response = await call_next(request)
except Exception:
logger.error("Request failed")
finally:
_request_id.reset()
return response
唯一需要注意的是,如果您使用多线程,multi-processing,或创建新的事件循环(在内部调用中),该块中的记录器实例将没有此 UUID。