logging模块详解

import logging 是 Python 内置的日志记录模块，用于替代 print() 进行更专业的日志管理。
基本用法

1. 最简单的配置

import logging

# 基本配置
logging.basicConfig(level=logging.INFO)
logging.info("这是一条信息")
logging.warning("这是一条警告")
logging.error("这是一条错误")

2. 不同日志级别

import logging

logging.debug("调试信息")     # 级别 10
logging.info("普通信息")      # 级别 20
logging.warning("警告信息")   # 级别 30
logging.error("错误信息")     # 级别 40
logging.critical("严重错误")  # 级别 50

常用配置

1. 配置输出格式

import logging

logging.basicConfig(
    level=logging.DEBUG,
    format='%(asctime)s - %(name)s - %(levelname)s - %(message)s',
    datefmt='%Y-%m-%d %H:%M:%S'
)

logging.info("程序启动")
# 输出: 2024-01-15 10:30:45 - root - INFO - 程序启动

2. 输出到文件

import logging

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

logging.info("这条会写入文件")

3. 同时输出到文件和控制台

import logging

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

# 文件 handler
file_handler = logging.FileHandler('app.log')
file_handler.setLevel(logging.INFO)

# 控制台 handler
console_handler = logging.StreamHandler()
console_handler.setLevel(logging.DEBUG)

# 设置格式
formatter = logging.Formatter('%(asctime)s - %(name)s - %(levelname)s - %(message)s')
file_handler.setFormatter(formatter)
console_handler.setFormatter(formatter)

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

# 使用
logger.debug("调试信息")
logger.info("普通信息")
logger.error("错误信息")

在函数/类中使用

import logging

def my_function():
    logger = logging.getLogger(__name__)
    logger.info("函数被调用")
    try:
        result = 10 / 0
    except ZeroDivisionError as e:
        logger.error(f"发生错误: {e}", exc_info=True)  # exc_info 打印完整堆栈

class MyClass:
    def __init__(self):
        self.logger = logging.getLogger(self.__class__.__name__)
    
    def do_something(self):
        self.logger.info("执行操作")

最佳实践

import logging

# 推荐的通用配置
logging.basicConfig(
    level=logging.INFO,
    format='%(asctime)s - %(levelname)s - %(pathname)s:%(lineno)d - %(message)s',
    handlers=[
        logging.FileHandler("app.log"),
        logging.StreamHandler()
    ]
)

# 在模块中
logger = logging.getLogger(__name__)

def main():
    logger.info("程序启动")
    # 你的代码
    logger.info("程序结束")

一、logging 模块架构
logging 模块采用模块化设计，包含四个主要组件：

Logger (日志器) → Handler (处理器) → Formatter (格式化器) → 输出
     ↓
Filter (过滤器) - 控制日志级别和内容

二、日志级别详解

import logging

# 预定义级别（数值越高越严重）
logging.NOTSET     = 0   # 未设置
logging.DEBUG      = 10  # 调试信息
logging.INFO       = 20  # 常规信息
logging.WARNING    = 30  # 警告信息
logging.ERROR      = 40  # 错误信息
logging.CRITICAL   = 50  # 严重错误

# 级别判断示例
logger = logging.getLogger('demo')
logger.setLevel(logging.WARNING)

logger.debug("调试")      # 不会输出（10 < 30）
logger.info("信息")       # 不会输出（20 < 30）
logger.warning("警告")    # 会输出（30 >= 30）
logger.error("错误")      # 会输出（40 >= 30）
logger.critical("严重")   # 会输出（50 >= 30）

三、Logger 对象详解
3.1 创建 Logger

import logging

# 创建根日志器（root logger）
root_logger = logging.getLogger()
root_logger.setLevel(logging.INFO)

# 创建命名日志器（推荐）
app_logger = logging.getLogger('myapp')
module_logger = logging.getLogger('myapp.module')

# 层级关系：myapp.module 继承 myapp 的设置
print(app_logger.parent)      # <RootLogger root>
print(module_logger.parent)   # <Logger myapp (INFO)>

3.2 Logger 的方法

import logging

logger = logging.getLogger(__name__)

# 基本日志方法
logger.debug('调试信息')
logger.info('普通信息')
logger.warning('警告信息')
logger.error('错误信息')
logger.critical('严重错误')
logger.exception('异常信息')  # 等同于 error，但会添加堆栈信息
logger.log(logging.INFO, '自定义级别的消息')

# 检查级别（避免不必要的计算）
if logger.isEnabledFor(logging.DEBUG):
    logger.debug(f'复杂计算: {expensive_function()}')

四、Handler 类型及使用
4.1 常用 Handler

import logging

# 1. StreamHandler - 输出到控制台
console_handler = logging.StreamHandler()
console_handler.setLevel(logging.DEBUG)

# 2. FileHandler - 输出到文件
file_handler = logging.FileHandler('app.log', encoding='utf-8')
file_handler.setLevel(logging.INFO)

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

# 4. TimedRotatingFileHandler - 按时间轮转
from logging.handlers import TimedRotatingFileHandler
time_handler = TimedRotatingFileHandler(
    'app.log',
    when='midnight',   # 每天午夜轮转
    interval=1,
    backupCount=30     # 保留30天
)

# 5. HTTPHandler - 发送到远程服务器
from logging.handlers import HTTPHandler
http_handler = HTTPHandler(
    host='localhost:8080',
    url='/log',
    method='POST'
)

# 6. SMTPHandler - 邮件发送错误
from logging.handlers import SMTPHandler
smtp_handler = SMTPHandler(
    mailhost=('smtp.gmail.com', 587),
    fromaddr='alert@myapp.com',
    toaddrs=['admin@myapp.com'],
    subject='应用错误',
    credentials=('user', 'password')
)
smtp_handler.setLevel(logging.ERROR)

4.2 完整的多 Handler 示例

import logging
from logging.handlers import RotatingFileHandler, TimedRotatingFileHandler

def setup_logger():
    """配置完整的日志系统"""
    
    # 创建主日志器
    logger = logging.getLogger('MyApp')
    logger.setLevel(logging.DEBUG)
    
    # 1. 控制台输出（仅 INFO 及以上）
    console = logging.StreamHandler()
    console.setLevel(logging.INFO)
    console_format = logging.Formatter('%(levelname)s - %(message)s')
    console.setFormatter(console_format)
    
    # 2. 文件输出（所有级别）
    file_handler = logging.FileHandler('app.log', encoding='utf-8')
    file_handler.setLevel(logging.DEBUG)
    file_format = logging.Formatter(
        '%(asctime)s - %(name)s - %(levelname)s - %(message)s',
        datefmt='%Y-%m-%d %H:%M:%S'
    )
    file_handler.setFormatter(file_format)
    
    # 3. 轮转文件（错误级别单独记录）
    error_handler = RotatingFileHandler(
        'error.log', 
        maxBytes=10*1024*1024,  # 10MB
        backupCount=10,
        encoding='utf-8'
    )
    error_handler.setLevel(logging.ERROR)
    error_format = logging.Formatter(
        '%(asctime)s - %(name)s - %(levelname)s - %(filename)s:%(lineno)d - %(message)s'
    )
    error_handler.setFormatter(error_format)
    
    # 添加所有处理器
    logger.addHandler(console)
    logger.addHandler(file_handler)
    logger.addHandler(error_handler)
    
    return logger

# 使用
logger = setup_logger()
logger.debug('调试信息')      # 仅文件
logger.info('普通信息')        # 控制台 + 文件
logger.error('错误信息')       # 控制台 + 文件 + 错误文件

五、Formatter 格式化详解
5.1 格式化字段

# 常用格式化字段
format_dict = {
    '%(name)s':        '日志器名称',
    '%(levelno)s':     '日志级别编号',
    '%(levelname)s':   '日志级别名称',
    '%(pathname)s':    '源文件完整路径',
    '%(filename)s':    '源文件名',
    '%(module)s':      '模块名',
    '%(lineno)d':      '行号',
    '%(funcName)s':    '函数名',
    '%(created)f':     '时间戳',
    '%(asctime)s':     '可读时间',
    '%(msecs)d':       '毫秒部分',
    '%(relativeCreated)d': '相对创建时间',
    '%(thread)d':      '线程ID',
    '%(threadName)s':  '线程名称',
    '%(process)d':     '进程ID',
    '%(message)s':     '日志消息'
}

# 实际示例
import logging

# 简单格式
simple_format = logging.Formatter('%(levelname)s: %(message)s')

# 详细格式
detailed_format = logging.Formatter(
    '%(asctime)s | %(levelname)-8s | %(filename)s:%(lineno)-4d | %(message)s',
    datefmt='%H:%M:%S'
)

# 生产环境格式
production_format = logging.Formatter(
    '%(asctime)s - %(name)s - %(levelname)s - %(process)d - %(threadName)s - %(message)s'
)

5.2 自定义 Formatter

import logging
import time

class ColoredFormatter(logging.Formatter):
    """彩色控制台输出格式器"""
    
    COLORS = {
        'DEBUG': '\033[36m',      # 青色
        'INFO': '\033[32m',       # 绿色
        'WARNING': '\033[33m',    # 黄色
        'ERROR': '\033[31m',      # 红色
        'CRITICAL': '\033[35m',   # 紫色
        'RESET': '\033[0m'        # 重置
    }
    
    def format(self, record):
        # 添加颜色
        color = self.COLORS.get(record.levelname, self.COLORS['RESET'])
        record.levelname = f"{color}{record.levelname}{self.COLORS['RESET']}"
        return super().format(record)

# 使用彩色格式器
logger = logging.getLogger()
console = logging.StreamHandler()
formatter = ColoredFormatter('%(levelname)s - %(message)s')
console.setFormatter(formatter)
logger.addHandler(console)
logger.setLevel(logging.DEBUG)

logger.debug('调试信息')    # 青色
logger.info('信息')         # 绿色
logger.warning('警告')      # 黄色
logger.error('错误')        # 红色

六、Filter 过滤器

import logging

class CustomFilter(logging.Filter):
    """自定义过滤器：只允许包含特定关键字的日志"""
    
    def __init__(self, keyword):
        super().__init__()
        self.keyword = keyword
    
    def filter(self, record):
        # 返回 True 表示允许，False 表示拒绝
        return self.keyword in record.getMessage()

# 使用过滤器
logger = logging.getLogger('app')
logger.setLevel(logging.DEBUG)

# 添加过滤器
filter_only_db = CustomFilter('database')
logger.addFilter(filter_only_db)

# 测试
logger.info('database connection')   # 会输出
logger.info('user login')             # 不会输出

# 移除过滤器
logger.removeFilter(filter_only_db)

七、配置方式
7.1 代码配置（最常用）

import logging

# 快速配置
logging.basicConfig(
    level=logging.INFO,
    format='%(asctime)s - %(levelname)s - %(message)s',
    handlers=[
        logging.FileHandler('app.log'),
        logging.StreamHandler()
    ]
)

7.2 字典配置（推荐生产环境）

import logging.config

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 - %(pathname)s:%(lineno)d - %(message)s'
        }
    },
    'handlers': {
        'console': {
            'class': 'logging.StreamHandler',
            'level': 'INFO',
            'formatter': 'standard',
            'stream': 'ext://sys.stdout'
        },
        'file': {
            'class': 'logging.handlers.RotatingFileHandler',
            'level': 'DEBUG',
            'formatter': 'detailed',
            'filename': 'app.log',
            'maxBytes': 10485760,  # 10MB
            'backupCount': 5,
            'encoding': 'utf-8'
        }
    },
    'loggers': {
        '': {  # root logger
            'handlers': ['console', 'file'],
            'level': 'DEBUG',
            'propagate': True
        },
        'myapp': {
            'handlers': ['file'],
            'level': 'INFO',
            'propagate': False
        }
    }
}

# 应用配置
logging.config.dictConfig(LOGGING_CONFIG)
logger = logging.getLogger('myapp')

7.3 文件配置

# logging.conf
[loggers]
keys=root,myapp

[handlers]
keys=consoleHandler,fileHandler

[formatters]
keys=simpleFormatter,detailedFormatter

[logger_root]
level=DEBUG
handlers=consoleHandler

[logger_myapp]
level=INFO
handlers=consoleHandler,fileHandler
qualname=myapp
propagate=0

[handler_consoleHandler]
class=StreamHandler
level=DEBUG
formatter=simpleFormatter
args=(sys.stdout,)

[handler_fileHandler]
class=logging.handlers.RotatingFileHandler
level=INFO
formatter=detailedFormatter
args=('app.log', 'a', 10485760, 5, 'utf-8')

[formatter_simpleFormatter]
format=%(asctime)s - %(levelname)s - %(message)s

[formatter_detailedFormatter]
format=%(asctime)s - %(name)s - %(levelname)s - %(filename)s:%(lineno)d - %(message)s

八、完整实战案例
案例1：Web应用日志系统

import logging
import logging.handlers
import time
from functools import wraps

class WebAppLogger:
    """Web应用日志管理器"""
    
    def __init__(self, app_name, log_dir='./logs'):
        self.app_name = app_name
        self.log_dir = log_dir
        self.logger = self._setup_logger()
    
    def _setup_logger(self):
        """配置日志系统"""
        import os
        os.makedirs(self.log_dir, exist_ok=True)
        
        logger = logging.getLogger(self.app_name)
        logger.setLevel(logging.DEBUG)
        
        # 访问日志
        access_handler = logging.handlers.TimedRotatingFileHandler(
            f'{self.log_dir}/access.log',
            when='midnight',
            interval=1,
            backupCount=30
        )
        access_format = logging.Formatter(
            '%(asctime)s - %(remote_ip)s - %(method)s - %(path)s - %(status)d - %(duration)dms'
        )
        access_handler.setFormatter(access_format)
        
        # 错误日志
        error_handler = logging.handlers.RotatingFileHandler(
            f'{self.log_dir}/error.log',
            maxBytes=10*1024*1024,
            backupCount=10
        )
        error_format = logging.Formatter(
            '%(asctime)s - %(levelname)s - %(pathname)s:%(lineno)d - %(message)s'
        )
        error_handler.setFormatter(error_format)
        error_handler.setLevel(logging.ERROR)
        
        # 控制台输出（开发环境）
        console = logging.StreamHandler()
        console.setLevel(logging.DEBUG)
        console_format = logging.Formatter(
            '%(asctime)s - %(name)s - %(levelname)s - %(message)s'
        )
        console.setFormatter(console_format)
        
        logger.addHandler(access_handler)
        logger.addHandler(error_handler)
        logger.addHandler(console)
        
        return logger
    
    def log_request(self, func):
        """记录请求的装饰器"""
        @wraps(func)
        def wrapper(*args, **kwargs):
            start_time = time.time()
            # 模拟获取请求信息
            remote_ip = kwargs.get('ip', 'unknown')
            method = kwargs.get('method', 'GET')
            path = kwargs.get('path', '/')
            
            try:
                result = func(*args, **kwargs)
                status = 200
                return result
            except Exception as e:
                status = 500
                self.logger.error(f"Request failed: {e}", exc_info=True)
                raise
            finally:
                duration = int((time.time() - start_time) * 1000)
                # 记录访问日志（使用额外参数）
                self.logger.info(
                    'Request processed',
                    extra={
                        'remote_ip': remote_ip,
                        'method': method,
                        'path': path,
                        'status': status,
                        'duration': duration
                    }
                )
        return wrapper

# 使用示例
web_logger = WebAppLogger('MyWebApp')

@web_logger.log_request
def handle_request(user_id, **request_info):
    """处理用户请求"""
    web_logger.logger.info(f"Processing user {user_id}")
    # 业务逻辑
    return {"user_id": user_id, "status": "success"}

# 模拟请求
handle_request(123, ip='192.168.1.1', method='GET', path='/api/user')

异步日志处理

import logging
import queue
from logging.handlers import QueueHandler, QueueListener
import threading
import time

class AsyncLogger:
    """异步日志处理器"""
    
    def __init__(self):
        self.log_queue = queue.Queue(-1)
        self.queue_handler = QueueHandler(self.log_queue)
        
        # 配置实际处理器
        self.file_handler = logging.FileHandler('async.log')
        self.file_handler.setFormatter(
            logging.Formatter('%(asctime)s - %(message)s')
        )
        
        # 创建监听器
        self.listener = QueueListener(
            self.log_queue, self.file_handler, respect_handler_level=True
        )
        
        # 配置根日志器
        root_logger = logging.getLogger()
        root_logger.addHandler(self.queue_handler)
        root_logger.setLevel(logging.DEBUG)
        
        # 启动监听器
        self.listener.start()
    
    def stop(self):
        self.listener.stop()

# 使用异步日志
async_logger = AsyncLogger()
logger = logging.getLogger()

# 日志记录不会阻塞主线程
for i in range(1000):
    logger.info(f"Message {i}")
    
# 程序结束时停止
async_logger.stop()