如何进行Python项目的日志管理？

logging

logging

logging.info("Something happened")

import logging
from logging.handlers import RotatingFileHandler
import os

# 定义日志文件路径
log_dir = "logs"
os.makedirs(log_dir, exist_ok=True)
log_file_path = os.path.join(log_dir, "my_application.log")

# 1. 获取Logger实例
logger = logging.getLogger(__name__)
logger.setLevel(logging.DEBUG) # 设置Logger的最低处理级别

# 2. 创建一个Formatter
formatter = logging.Formatter(
    '%(asctime)s - %(name)s - %(levelname)s - %(funcName)s - %(lineno)d - %(message)s'
)

# 3. 创建StreamHandler（输出到控制台）
console_handler = logging.StreamHandler()
console_handler.setLevel(logging.INFO) # 控制台只显示INFO及以上级别
console_handler.setFormatter(formatter)

# 4. 创建RotatingFileHandler（输出到文件，文件大小限制10MB，保留5个备份）
file_handler = RotatingFileHandler(
    log_file_path,
    maxBytes=10 * 1024 * 1024, # 10 MB
    backupCount=5,
    encoding='utf-8'
)
file_handler.setLevel(logging.DEBUG) # 文件记录所有DEBUG及以上级别
file_handler.setFormatter(formatter)

# 5. 将Handler添加到Logger
logger.addHandler(console_handler)
logger.addHandler(file_handler)

# 示例使用
logger.debug("这是一个调试信息")
logger.info("程序启动，加载配置...")
logger.warning("发现潜在问题：配置项缺失")
try:
    1 / 0
except ZeroDivisionError:
    logger.error("发生严重错误：除零异常", exc_info=True) # exc_info=True 会记录异常堆栈

logging.config.dictConfig

dictConfig

logging.config.dictConfig

import logging.config
import yaml # 或者 json

# 假设这是一个 YAML 配置文件内容
LOGGING_CONFIG = {
    'version': 1,
    'disable_existing_loggers': False, # 禁用已存在的logger，通常设为False
    'formatters': {
        'standard': {
            'format': '%(asctime)s - %(name)s - %(levelname)s - %(message)s'
        },
        'verbose': {
            'format': '%(asctime)s - %(name)s - %(levelname)s - %(filename)s:%(lineno)d - %(funcName)s - %(message)s'
        }
    },
    'handlers': {
        'console': {
            'class': 'logging.StreamHandler',
            'level': 'INFO',
            'formatter': 'standard',
            'stream': 'ext://sys.stdout'
        },
        'file_dev': {
            'class': 'logging.handlers.RotatingFileHandler',
            'level': 'DEBUG',
            'formatter': 'verbose',
            'filename': 'logs/app_dev.log',
            'maxBytes': 1048576, # 1MB
            'backupCount': 3,
        },
        'file_prod': {
            'class': 'logging.handlers.RotatingFileHandler',
            'level': 'INFO',
            'formatter': 'standard',
            'filename': 'logs/app_prod.log',
            'maxBytes': 10485760, # 10MB
            'backupCount': 5,
        }
    },
    'loggers': {
        'my_app': { # 你的应用主Logger
            'handlers': ['console', 'file_dev'], # 开发环境使用
            'level': 'DEBUG',
            'propagate': False # 不向父级Logger传递日志
        },
        'another_module': {
            'handlers': ['file_prod'],
            'level': 'WARNING',
            'propagate': False
        }
    },
    'root': { # 根Logger，通常作为所有未显式配置Logger的兜底
        'handlers': ['console'],
        'level': 'WARNING'
    }
}

# 在应用启动时加载配置
# logging.config.dictConfig(LOGGING_CONFIG)

# 动态切换环境的例子
ENV = os.environ.get('APP_ENV', 'development') # 从环境变量获取当前环境

if ENV == 'production':
    LOGGING_CONFIG['loggers']['my_app']['handlers'] = ['console', 'file_prod']
    LOGGING_CONFIG['loggers']['my_app']['level'] = 'INFO'
    # 生产环境可能还需要额外的Handler，比如发送到中央日志系统
    # LOGGING_CONFIG['handlers']['remote'] = {...}
    # LOGGING_CONFIG['loggers']['my_app']['handlers'].append('remote')
elif ENV == 'development':
    LOGGING_CONFIG['loggers']['my_app']['handlers'] = ['console', 'file_dev']
    LOGGING_CONFIG['loggers']['my_app']['level'] = 'DEBUG'

logging.config.dictConfig(LOGGING_CONFIG)
logger = logging.getLogger('my_app')

logger.debug("这条消息只在开发环境显示")
logger.info("这条消息在所有环境都显示")

APP_ENV

python-json-logger

logging.Formatter

import logging
import logging.config
import json_log_formatter # pip install python-json-logger

# 定义一个自定义的JSON Formatter
class CustomJsonFormatter(json_log_formatter.JsonFormatter):
    def add_fields(self, log_record, message_dict):
        super(CustomJsonFormatter, self).add_fields(log_record, message_dict)
        # 添加自定义字段
        message_dict['service'] = 'my_awesome_service'
        message_dict['version'] = '1.0.0'
        # 如果有request_id或user_id，也可以从线程局部存储中获取
        # message_dict['request_id'] = getattr(threading.current_thread(), 'request_id', 'N/A')

# 示例配置
LOGGING_CONFIG_JSON = {
    'version': 1,
    'disable_existing_loggers': False,
    'formatters': {
        'json': {
            '()': CustomJsonFormatter, # 使用自定义的JSON Formatter
            'format': '(levelname) (name) (message)' # 这个format字符串实际上会被json_log_formatter忽略，但必须存在
        }
    },
    'handlers': {
        'json_console': {
            'class': 'logging.StreamHandler',
            'level': 'INFO',
            'formatter': 'json',
            'stream': 'ext://sys.stdout'
        },
        'json_file': {
            'class': 'logging.handlers.RotatingFileHandler',
            'level': 'DEBUG',
            'formatter': 'json',
            'filename': 'logs/app_structured.log',
            'maxBytes': 1048576,
            'backupCount': 3,
        }
    },
    'loggers': {
        'my_app_json': {
            'handlers': ['json_console', 'json_file'],
            'level': 'DEBUG',
            'propagate': False
        }
    },
    'root': {
        'handlers': ['json_console'],
        'level': 'WARNING'
    }
}

logging.config.dictConfig(LOGGING_CONFIG_JSON)
json_logger = logging.getLogger('my_app_json')

json_logger.info("用户登录成功", extra={'user_id': 123, 'ip_address': '192.168.1.100'})
json_logger.warning("数据库连接超时", extra={'db_host': 'localhost', 'port': 5432})
try:
    result = 1 / 0
except ZeroDivisionError:
    json_logger.error("计算失败", exc_info=True, extra={'operation': 'division'})

extra