如何基于loguru按执行组拆分日志到独立文件并配置轮转

1. 统一配置流水线日志规则 #

先在项目的公共配置文件（比如log_config.py）里，为每个流水线单独配置日志输出规则，包括专属文件路径、独立轮转策略和过滤条件，确保所有模块共用同一套日志配置。

示例代码：

from loguru import logger
import os

logs_directory = "./logs"
os.makedirs(logs_directory, exist_ok=True)

def setup_pipeline_logs():
    # Store1销售流水线日志
    logger.add(
        sink=os.path.join(logs_directory, "store1_sales.json"),
        level="INFO",
        rotation="30 day",  # 独立轮转规则
        serialize=True,
        encoding="utf-8",
        filter=lambda record: record["extra"].get("pipeline") == "store1_sales"
    )
    
    # Store1仓库流水线日志
    logger.add(
        sink=os.path.join(logs_directory, "store1_warehouses.json"),
        level="INFO",
        rotation="1 week",  # 不同轮转周期
        serialize=True,
        encoding="utf-8",
        filter=lambda record: record["extra"].get("pipeline") == "store1_warehouses"
    )
    
    # Store2销售流水线日志
    logger.add(
        sink=os.path.join(logs_directory, "store2_sales.json"),
        level="INFO",
        rotation="1 month",
        serialize=True,
        encoding="utf-8",
        filter=lambda record: record["extra"].get("pipeline") == "store2_sales"
    )

# 初始化日志配置
setup_pipeline_logs()

2. 在各模块的流水线函数中绑定标记 #

不同目录的处理脚本里，通过logger.bind()给当前函数的日志打上对应流水线的标记，日志会自动被匹配的过滤器路由到指定文件。

抽取模块（示例：store1/extract_sales.py）： #

from log_config import logger

def store1_extract_sales():
    # 绑定store1_sales标记
    pipe_logger = logger.bind(pipeline="store1_sales")
    pipe_logger.info("开始抽取store1销售数据")
    # 业务逻辑代码...
    pipe_logger.success("store1销售数据抽取完成")

转换模块（示例：store1/convert_sales.py）： #

from log_config import logger

def store1_convert_sales():
    pipe_logger = logger.bind(pipeline="store1_sales")
    pipe_logger.info("开始转换store1销售数据")
    # 业务逻辑代码...
    pipe_logger.warning("转换过程中发现3条无效数据")

加载模块（示例：store2/load_sales.py）： #

from log_config import logger

def store2_load_sales():
    pipe_logger = logger.bind(pipeline="store2_sales")
    pipe_logger.info("开始加载store2销售数据到数据库")
    # 业务逻辑代码...
    pipe_logger.error("数据库连接超时，加载失败")

3. 核心逻辑说明 #

• 过滤器匹配：每个日志sink通过filter参数判断日志记录的extra.pipeline标记，匹配成功则写入对应文件。
• 独立轮转：每个logger.add()的rotation参数可单独设置，各流水线日志轮转完全独立。
• 跨目录复用：所有脚本导入同一配置文件，避免重复配置，保证规则一致性。
• 全量日志可选：如果需要保留一份完整日志，只需额外添加一个不带过滤器的sink即可。

常见问题排查 #

• 日志未写入指定文件：检查bind的标记字符串和过滤器中的值是否完全一致（注意大小写、拼写）。
• 轮转规则不生效：确认rotation参数格式正确，支持"30 day"、"1 week"、"100 MB"等格式。

内容的提问来源于stack exchange，提问作者Sergey Bakaev Rettley