多模块Python日志配置问题（对接Azure Monitor）

场景与需求 #

• 场景：FastAPI应用，Docker容器部署，日志输出至Azure Monitor/Application Insights
• 核心需求：
  1. 集中一处定义日志处理器、格式化器与过滤器
  2. INFO及以下级别日志发送至处理器A，INFO以上级别发送至处理器B
  3. 符合最佳实践的项目结构

当前项目结构 #

dockerfile
main.py
requirements.txt
logger/
    logger.py
some_other_module/
    module.py

现有代码 #

logger.py #

import logging
from logging import LogRecord
from datetime import datetime
from pytz import timezone
from opencensus.ext.azure.log_exporter import AzureEventHandler

def log_warning_or_below(record : LogRecord):
    if(record.levelname in [logging.DEBUG, logging.INFO, logging.WARN, logging.WARNING]):
            return True
    return False

def build_custom_events_handler():
    connection_string='InstrumentationKey=xxxxxxxxxxxxxxxxxxxx'
    handler=AzureEventHandler(connection_string=connection_string)
    handler.setLevel(logging.INFO)
    handler.addFilter(log_warning_or_below)

    return handler

def init():
    custom_events_handler = build_custom_events_handler()
    logging.basicConfig(
        format='%(asctime)s - %(message)s',
        level=logging.INFO,
        datefmt='%d-%m-%Y %H:%M:%S',
        handlers=[custom_events_handler]
    )
    logging.Formatter.converter = timetz

init()

main.py #

....various imports...
import logger.logger

logger = logger.logger.getLogger('main')
logger.info('Start of main')

def _create_app() -> FastAPI:
    api = FastAPI(
        title='my title',
        description="some desc",
        version='1.0',
    )
    return api

app = _create_app()

@app.post('/myroute')
def do_something():
    logger.info('inside the function')
    return 'my route was hit'

问题现状 #

部署Docker容器后，FastAPI默认日志可正常显示在Azure中，但自定义日志器输出的日志完全无法展示。同时对以下内容存在疑问：

• 根日志器的配置是否会被子日志器继承？
• 使用logging.getLogger(__name__)的最佳实践是否因目录结构问题无法生效？

已尝试方案：

1. 在logger.py中创建静态日志器（可正常工作，但不符合“每个模块一个日志器”的需求）
2. Jupyter Notebook中测试代码可正常输出日志，但部署后失效
3. 尝试使用DictConfig，复杂度超出预期
4. 查阅多模块日志配置相关内容，未找到适配场景的解决方案

1. 过滤器逻辑错误（核心问题） #

你的log_warning_or_below函数将数字类型的日志级别常量（如logging.DEBUG=10）与字符串类型的record.levelname（如'DEBUG'）做对比，导致判断永远不成立，所有日志被过滤器拦截。正确写法：

def log_warning_or_below(record: LogRecord):
    return record.levelno <= logging.WARNING

2. 未定义timetz函数导致初始化失败 #

logging.Formatter.converter = timetz引用了未定义的函数，会直接抛出NameError，导致日志配置初始化失败。如需时区转换，补充定义函数：

def timetz(*args):
    return datetime.now(timezone('UTC')).timetuple()

无需时区转换则直接删除该行代码。

3. 日志器获取方式错误 #

main.py中获取日志器的方式有误，应直接使用标准库logging模块获取，而非从logger.logger模块引用：

# main.py修改后
import logging
# 导入logger模块触发日志初始化
import logger.logger

# 最佳实践：用模块名作为日志器名称
logger = logging.getLogger(__name__)

该方式获取的日志器会自动继承根日志器的配置，与目录结构无关。

4. 处理器级别与过滤器配合优化 #

若要实现INFO及以下→处理器A，INFO以上→处理器B的分流，需调整配置逻辑：

• 处理器A：设置级别为logging.DEBUG（先放行所有DEBUG+日志），用过滤器保留record.levelno <= logging.INFO
• 处理器B：设置级别为logging.WARNING（直接过滤WARNING以下日志），无需额外过滤器（可选添加）

调整后项目结构 #

新增空文件logger/__init__.py，让logger成为Python可识别的包：

dockerfile
main.py
requirements.txt
logger/
    __init__.py
    logger.py
some_other_module/
    module.py

优化后的logger.py #

import logging
from logging import LogRecord
from datetime import datetime
from pytz import timezone
from opencensus.ext.azure.log_exporter import AzureEventHandler

# 时区转换函数（按需启用）
def timetz(*args):
    return datetime.now(timezone('UTC')).timetuple()

# 过滤器：保留INFO及以下级别
def filter_info_and_below(record: LogRecord):
    return record.levelno <= logging.INFO

# 过滤器：保留WARNING及以上级别
def filter_above_info(record: LogRecord):
    return record.levelno > logging.INFO

def build_handler_a():
    """处理器A：处理INFO及以下日志"""
    connection_string = 'InstrumentationKey=xxxxxxxxxxxxxxxxxxxx'
    handler = AzureEventHandler(connection_string=connection_string)
    handler.setLevel(logging.DEBUG)
    handler.addFilter(filter_info_and_below)
    # 配置格式化器
    formatter = logging.Formatter(
        '%(asctime)s - %(name)s - %(levelname)s - %(message)s',
        datefmt='%d-%m-%Y %H:%M:%S'
    )
    handler.setFormatter(formatter)
    return handler

def build_handler_b():
    """处理器B：处理WARNING及以上日志"""
    connection_string = 'InstrumentationKey=xxxxxxxxxxxxxxxxxxxx'
    handler = AzureEventHandler(connection_string=connection_string)
    handler.setLevel(logging.WARNING)
    # 配置格式化器（可添加异常信息）
    formatter = logging.Formatter(
        '%(asctime)s - %(name)s - %(levelname)s - %(message)s - %(exc_info)s',
        datefmt='%d-%m-%Y %H:%M:%S'
    )
    handler.setFormatter(formatter)
    return handler

def init():
    root_logger = logging.getLogger()
    # 清空默认处理器，避免重复输出
    root_logger.handlers.clear()
    # 根日志器设为DEBUG级别，确保所有子日志器日志可传递
    root_logger.setLevel(logging.DEBUG)
    
    # 添加两个处理器
    root_logger.addHandler(build_handler_a())
    root_logger.addHandler(build_handler_b())
    
    # 启用时区转换（按需启用）
    logging.Formatter.converter = timetz

# 自动执行初始化
init()

优化后的main.py #

from fastapi import FastAPI
import logging
# 导入logger包触发日志初始化
import logger.logger

# 使用模块名作为日志器名称，符合最佳实践
logger = logging.getLogger(__name__)

logger.info('Start of main')

def _create_app() -> FastAPI:
    api = FastAPI(
        title='my title',
        description="some desc",
        version='1.0',
    )
    return api

app = _create_app()

@app.post('/myroute')
def do_something():
    logger.info('inside the function')
    logger.warning('这是一条警告日志')
    return 'my route was hit'

其他模块日志使用示例（some_other_module/module.py） #

import logging

logger = logging.getLogger(__name__)

def some_function():
    logger.debug('这是一条调试日志')
    logger.error('这是一条错误日志')

1. 根日志器必须设置为logging.DEBUG，确保所有子日志器的不同级别日志都能传递到根日志器的处理器。
2. 处理器级别与过滤器配合使用，实现精准的日志分流。
3. logging.getLogger(__name__)是通用最佳实践，每个模块的日志器名称唯一，便于排查问题，且自动继承根日志器配置，与目录结构无关。
4. Docker部署时，确保容器内安装所有依赖（如opencensus-ext-azure），建议用环境变量存储Azure连接字符串，避免硬编码。

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