如何为Python异步代码配置日志，实现多任务日志流区分与跨模块传递？

核心思路 #

利用Python的contextvars存储每个异步任务的唯一ID，结合日志过滤器自动将ID注入所有日志记录，无需修改原有模块的日志调用逻辑，同时兼容异步协程的上下文切换。

具体实现步骤 #

1. 定义全局上下文变量与日志过滤器 #

创建一个工具模块（如log_utils.py），统一管理任务ID的上下文存储和日志过滤逻辑：

import contextvars
import logging
import uuid

# 上下文变量，用于存储当前任务的唯一ID，默认值为UNKNOWN
task_id_var = contextvars.ContextVar("task_id", default="UNKNOWN")

class TaskIDFilter(logging.Filter):
    def filter(self, record):
        # 从上下文获取当前任务ID，注入日志记录的字段中
        record.task_id = task_id_var.get()
        return True

2. 全局配置日志系统 #

在程序入口处配置日志格式，并给所有日志器添加自定义过滤器：

import logging
from log_utils import TaskIDFilter

# 配置日志格式，将task_id作为固定前缀加入日志内容
logging.basicConfig(
    level=logging.INFO,
    format="%(asctime)s [%(task_id)s] %(name)s - %(levelname)s: %(message)s"
)

# 给所有已注册的日志器添加过滤器，包括第三方模块的日志器
for logger_name in logging.root.manager.loggerDict:
    logger = logging.getLogger(logger_name)
    logger.addFilter(TaskIDFilter())
# 确保根日志器也应用过滤器
logging.root.addFilter(TaskIDFilter())

3. 异步任务中绑定唯一ID #

在每个异步任务启动时生成UUID，设置到上下文变量中，确保任务执行全程的日志都能关联该ID：

import asyncio
from log_utils import task_id_var, uuid
import module_b  # 被调用的业务模块

async def execute_task():
    # 生成并绑定当前任务的唯一ID
    task_id = str(uuid.uuid4())
    task_id_var.set(task_id)
    
    logger = logging.getLogger(__name__)
    logger.info("任务启动，开始执行业务逻辑")
    
    # 调用其他模块的函数，该模块的日志会自动带上当前任务ID
    await module_b.process_data()
    
    logger.info("任务执行完成")

async def main():
    # 并发启动多个任务测试日志隔离效果
    await asyncio.gather(
        execute_task(),
        execute_task(),
        execute_task()
    )

if __name__ == "__main__":
    asyncio.run(main())

4. 被调用模块保持原有日志写法 #

业务模块（如module_b.py）无需修改日志初始化代码，正常使用标准日志接口即可：

import logging

logger = logging.getLogger(__name__)
logger.setLevel(logging.INFO)

async def process_data():
    logger.info("开始处理数据")
    # 模拟异步业务操作
    await asyncio.sleep(0.5)
    logger.info("数据处理完成，返回结果")

关键细节说明 #

• contextvars是Python 3.7+专为异步协程设计的上下文管理工具，能正确处理协程切换时的上下文传递，比threading.local更适配asyncio场景。
• 自定义过滤器自动注入任务ID，无需在日志调用时手动传入参数，完全兼容原有代码的日志写法。
• 全局添加过滤器可覆盖所有日志器，包括第三方依赖模块的日志，确保全链路日志都带有任务ID。

内容的提问来源于stack exchange，提问作者persson