Python 日志系统设计：构建可观测的应用

第一程序员

325人浏览 · 2026-05-08 19:57:19

第一程序员 · 2026-05-08 19:57:19 发布

Python 日志系统设计：构建可观测的应用

引言

大家好，我是一名正在从Rust转向Python的后端开发者。在构建大型应用程序时，日志系统是不可或缺的组成部分。良好的日志系统不仅能帮助我们追踪问题，还能提供应用运行时的关键洞察。作为从Rust过来的开发者，我发现Python的logging模块功能非常强大，但配置起来也相对复杂。今天，我想和大家分享一下我在设计Python日志系统方面的一些经验。

日志系统的重要性

为什么需要日志系统？

问题排查：快速定位和解决生产环境中的问题
性能监控：追踪应用性能指标
安全审计：记录关键操作和访问记录
业务分析：通过日志数据了解用户行为

日志级别

Python的logging模块定义了以下日志级别：

级别	数值	用途
`DEBUG`	10	详细的调试信息
`INFO`	20	一般信息
`WARNING`	30	警告信息
`ERROR`	40	错误信息
`CRITICAL`	50	严重错误信息

基础配置

简单配置

import logging

# 基础配置
logging.basicConfig(
    level=logging.INFO,
    format='%(asctime)s - %(name)s - %(levelname)s - %(message)s'
)

logger = logging.getLogger(__name__)

logger.debug('这是一条调试信息')
logger.info('这是一条普通信息')
logger.warning('这是一条警告信息')
logger.error('这是一条错误信息')
logger.critical('这是一条严重错误信息')

输出到文件

import logging

logging.basicConfig(
    level=logging.DEBUG,
    format='%(asctime)s - %(name)s - %(levelname)s - %(message)s',
    filename='app.log',
    filemode='w'  # 'w' 覆盖模式，'a' 追加模式
)

logger = logging.getLogger(__name__)
logger.info('应用启动')

高级配置

使用配置字典

import logging
from logging.config import dictConfig

logging_config = {
    'version': 1,
    'disable_existing_loggers': False,
    'formatters': {
        'standard': {
            'format': '%(asctime)s - %(name)s - %(levelname)s - %(message)s'
        },
        'detailed': {
            'format': '%(asctime)s - %(name)s - %(levelname)s - %(module)s:%(lineno)d - %(message)s'
        }
    },
    'handlers': {
        'console': {
            'class': 'logging.StreamHandler',
            'formatter': 'standard',
            'level': 'INFO'
        },
        'file': {
            'class': 'logging.FileHandler',
            'filename': 'app.log',
            'formatter': 'detailed',
            'level': 'DEBUG'
        },
        'error_file': {
            'class': 'logging.FileHandler',
            'filename': 'error.log',
            'formatter': 'detailed',
            'level': 'ERROR'
        }
    },
    'loggers': {
        '': {  # root logger
            'handlers': ['console', 'file', 'error_file'],
            'level': 'DEBUG',
            'propagate': True
        }
    }
}

dictConfig(logging_config)
logger = logging.getLogger(__name__)
logger.debug('调试信息')
logger.info('普通信息')
logger.error('错误信息')

自定义日志格式

import logging

# 创建logger
logger = logging.getLogger('my_app')
logger.setLevel(logging.DEBUG)

# 创建formatter
formatter = logging.Formatter(
    '%(asctime)s - %(name)s - %(levelname)s - %(process)d - %(thread)d - %(message)s'
)

# 创建handler
console_handler = logging.StreamHandler()
console_handler.setLevel(logging.INFO)
console_handler.setFormatter(formatter)

file_handler = logging.FileHandler('app.log')
file_handler.setLevel(logging.DEBUG)
file_handler.setFormatter(formatter)

# 添加handler到logger
logger.addHandler(console_handler)
logger.addHandler(file_handler)

logger.info('应用启动')

实际应用场景

场景1：多模块日志管理

在大型项目中，每个模块应该有自己的logger：

# module_a.py
import logging

logger = logging.getLogger(__name__)

def do_something():
    logger.debug('正在执行do_something')
    try:
        # 业务逻辑
        logger.info('操作成功')
    except Exception as e:
        logger.error(f'操作失败: {e}', exc_info=True)

# main.py
import logging
from logging.config import dictConfig
import module_a

logging_config = {
    'version': 1,
    'disable_existing_loggers': False,
    'formatters': {
        'standard': {
            'format': '%(asctime)s - %(name)s - %(levelname)s - %(message)s'
        }
    },
    'handlers': {
        'console': {
            'class': 'logging.StreamHandler',
            'formatter': 'standard',
            'level': 'INFO'
        }
    },
    'loggers': {
        'module_a': {
            'handlers': ['console'],
            'level': 'DEBUG',
            'propagate': False
        }
    }
}

dictConfig(logging_config)

if __name__ == '__main__':
    logger = logging.getLogger(__name__)
    logger.info('应用启动')
    module_a.do_something()

场景2：日志轮转

对于长期运行的应用，日志文件会不断增大，需要进行轮转：

import logging
from logging.handlers import RotatingFileHandler, TimedRotatingFileHandler

logger = logging.getLogger('my_app')
logger.setLevel(logging.DEBUG)

# 按文件大小轮转
rotating_handler = RotatingFileHandler(
    'app.log',
    maxBytes=1024 * 1024 * 5,  # 5MB
    backupCount=5  # 保留5个备份
)

# 按时间轮转
timed_handler = TimedRotatingFileHandler(
    'app.log',
    when='midnight',  # 每天午夜轮转
    interval=1,
    backupCount=7  # 保留7天的日志
)

formatter = logging.Formatter('%(asctime)s - %(name)s - %(levelname)s - %(message)s')
rotating_handler.setFormatter(formatter)
timed_handler.setFormatter(formatter)

logger.addHandler(rotating_handler)
logger.addHandler(timed_handler)

场景3：结构化日志

在现代应用中，结构化日志（JSON格式）越来越流行：

import logging
import json
from datetime import datetime

class JsonFormatter(logging.Formatter):
    def format(self, record):
        log_record = {
            'timestamp': datetime.now().isoformat(),
            'logger': record.name,
            'level': record.levelname,
            'message': record.getMessage(),
            'module': record.module,
            'line': record.lineno
        }
        
        if record.exc_info:
            log_record['exception'] = self.formatException(record.exc_info)
        
        return json.dumps(log_record)

logger = logging.getLogger('structured_logger')
logger.setLevel(logging.DEBUG)

handler = logging.StreamHandler()
handler.setFormatter(JsonFormatter())
logger.addHandler(handler)

logger.info('用户登录', extra={'user_id': 123, 'ip': '192.168.1.1'})
logger.error('数据库连接失败', extra={'db_host': 'localhost', 'db_port': 5432})

场景4：日志过滤

根据条件过滤日志：

import logging

class ErrorFilter(logging.Filter):
    def filter(self, record):
        # 只允许ERROR级别及以上的日志通过
        return record.levelno >= logging.ERROR

logger = logging.getLogger('filtered_logger')
logger.setLevel(logging.DEBUG)

console_handler = logging.StreamHandler()
console_handler.setLevel(logging.DEBUG)
console_handler.addFilter(ErrorFilter())

formatter = logging.Formatter('%(asctime)s - %(levelname)s - %(message)s')
console_handler.setFormatter(formatter)

logger.addHandler(console_handler)

logger.debug('调试信息')  # 不会输出
logger.info('普通信息')   # 不会输出
logger.error('错误信息')  # 会输出

最佳实践

1. 使用结构化日志

结构化日志便于日志分析和查询：

import logging

logger = logging.getLogger(__name__)

# 使用extra参数添加额外信息
logger.info('用户登录成功', extra={
    'user_id': 123,
    'username': 'john_doe',
    'ip_address': '192.168.1.100',
    'user_agent': 'Mozilla/5.0'
})

2. 记录异常信息

try:
    # 可能出错的代码
    result = risky_operation()
except Exception as e:
    # 记录完整的异常堆栈
    logger.error(f'操作失败: {e}', exc_info=True)
    # 或者使用logger.exception自动记录堆栈
    logger.exception('操作失败')

3. 使用日志包装器

import logging
from functools import wraps

def log_function_call(func):
    @wraps(func)
    def wrapper(*args, **kwargs):
        logger = logging.getLogger(func.__module__)
        logger.debug(f'调用 {func.__name__}, 参数: args={args}, kwargs={kwargs}')
        try:
            result = func(*args, **kwargs)
            logger.debug(f'{func.__name__} 执行成功, 返回: {result}')
            return result
        except Exception as e:
            logger.error(f'{func.__name__} 执行失败: {e}', exc_info=True)
            raise
    return wrapper

@log_function_call
def process_data(data):
    # 处理数据
    return data

4. 配置分离

将日志配置放在单独的配置文件中：

# logging_config.py
LOGGING_CONFIG = {
    'version': 1,
    'disable_existing_loggers': False,
    'formatters': {
        'standard': {
            'format': '%(asctime)s - %(name)s - %(levelname)s - %(message)s'
        }
    },
    'handlers': {
        'console': {
            'class': 'logging.StreamHandler',
            'formatter': 'standard',
            'level': 'INFO'
        }
    },
    'loggers': {
        '': {
            'handlers': ['console'],
            'level': 'DEBUG',
            'propagate': True
        }
    }
}

# main.py
from logging.config import dictConfig
from logging_config import LOGGING_CONFIG

dictConfig(LOGGING_CONFIG)

与Rust日志系统的对比

特性	Python `logging`	Rust `log` crate
配置方式	代码配置或配置文件	编译时配置
日志级别	运行时可调整	编译时确定
结构化日志	需要自定义formatter	原生支持
性能	一般	非常快
生态	丰富的第三方库	简洁的核心库

实战项目：完整的日志系统

import logging
from logging.config import dictConfig
from logging.handlers import RotatingFileHandler
import json
from datetime import datetime

class JsonFormatter(logging.Formatter):
    def format(self, record):
        log_record = {
            'timestamp': datetime.now().isoformat(),
            'logger': record.name,
            'level': record.levelname,
            'message': record.getMessage(),
            'module': record.module,
            'line': record.lineno,
            'process': record.process,
            'thread': record.thread
        }
        
        if record.exc_info:
            log_record['exception'] = self.formatException(record.exc_info)
        
        if hasattr(record, 'extra'):
            log_record.update(record.extra)
        
        return json.dumps(log_record)

def setup_logging():
    logging_config = {
        'version': 1,
        'disable_existing_loggers': False,
        'formatters': {
            'json': {
                '()': JsonFormatter
            },
            'standard': {
                'format': '%(asctime)s - %(name)s - %(levelname)s - %(message)s'
            }
        },
        'handlers': {
            'console': {
                'class': 'logging.StreamHandler',
                'formatter': 'standard',
                'level': 'INFO'
            },
            'file': {
                'class': 'logging.handlers.RotatingFileHandler',
                'filename': 'app.log',
                'formatter': 'json',
                'level': 'DEBUG',
                'maxBytes': 1024 * 1024 * 5,
                'backupCount': 5
            },
            'error_file': {
                'class': 'logging.handlers.RotatingFileHandler',
                'filename': 'error.log',
                'formatter': 'json',
                'level': 'ERROR',
                'maxBytes': 1024 * 1024 * 5,
                'backupCount': 5
            }
        },
        'loggers': {
            'app': {
                'handlers': ['console', 'file', 'error_file'],
                'level': 'DEBUG',
                'propagate': False
            },
            'app.database': {
                'handlers': ['file'],
                'level': 'DEBUG',
                'propagate': False
            },
            'app.api': {
                'handlers': ['console', 'file'],
                'level': 'INFO',
                'propagate': False
            }
        }
    }
    
    dictConfig(logging_config)

if __name__ == '__main__':
    setup_logging()
    
    logger = logging.getLogger('app')
    db_logger = logging.getLogger('app.database')
    api_logger = logging.getLogger('app.api')
    
    logger.info('应用启动')
    db_logger.debug('连接数据库')
    api_logger.info('处理API请求')

总结

构建一个良好的日志系统是构建可观测应用的关键。通过合理配置日志级别、输出格式和存储策略，我们可以实现：

问题快速定位：通过详细的日志信息快速定位问题
性能监控：通过日志分析应用性能
安全审计：记录关键操作
业务分析：通过日志数据了解系统运行状态

作为从Rust转向Python的开发者，我发现Python的日志系统虽然配置复杂，但灵活性很高。通过合理使用，可以构建出满足各种需求的日志系统。

延伸阅读：

亚马逊云科技技术品牌专区

更多推荐

CMU 11-785 深度学习导论笔记（一）

神经网络是人工智能中的一种方法，它教会计算机以受人类大脑启发的方式处理数据。近年来，它已成为各种模式识别、预测和分析问题的主要研究方向之一。神经网络在许多问题上确立了最先进的技术水平，并且常常大幅超越之前的基准。上一节我们介绍了神经网络的基本定义，本节中我们来看看神经网络带来的一些突破性应用。语音助手：例如 Siri、Alexa、Google Assistant。视觉与感知：例如人脸检测、人脸识别