Loguru

概述

Loguru 是一个功能强大且易于使用的 Python 日志库，旨在简化日志记录的复杂性。它提供了开箱即用的功能，无需复杂配置即可开始使用。

安装

pip install loguru

基础使用

1. 导入和基本使用

from loguru import logger

# 基本日志记录
logger.debug("这是一个调试信息")
logger.info("这是一个信息日志")
logger.warning("这是一个警告")
logger.error("这是一个错误")
logger.critical("这是一个严重错误")

2. 自动异常捕获

from loguru import logger

@logger.catch
def my_function(x, y, z):
    """
    带有自动异常捕获的函数
    :param x: 参数x
    :param y: 参数y
    :param z: 参数z
    :return: 计算结果
    """
    return 1 / (x + y + z)

# 或者使用装饰器参数
@logger.catch(reraise=True)
def risky_function():
    """
    有风险的函数，捕获异常后重新抛出
    :return: None
    """
    raise ValueError("出错了！")

高级配置

1. 添加自定义处理器

from loguru import logger
import sys

# 移除默认处理器
logger.remove()

# 添加控制台输出
logger.add(
    sys.stderr,
    format="<green>{time:YYYY-MM-DD HH:mm:ss.SSS}</green> | <level>{level: <8}</level> | <cyan>{name}</cyan>:<cyan>{function}</cyan>:<cyan>{line}</cyan> - <level>{message}</level>",
    level="INFO"
)

# 添加文件输出
logger.add(
    "logs/app_{time:YYYY-MM-DD}.log",
    rotation="00:00",  # 每天午夜轮转
    retention="30 days",  # 保留30天
    compression="zip",  # 压缩旧日志
    level="DEBUG",
    format="{time:YYYY-MM-DD HH:mm:ss.SSS} | {level: <8} | {name}:{function}:{line} - {message}"
)

2. 日志轮转配置

from loguru import logger

# 按大小轮转
logger.add("logs/app.log", rotation="100 MB")

# 按时间轮转
logger.add("logs/app.log", rotation="1 week")

# 按数量轮转
logger.add("logs/app.log", rotation="10")

# 自定义轮转条件
def should_rotate(message, file):
    """
    自定义轮转条件
    :param message: 日志消息
    :param file: 文件对象
    :return: 是否需要轮转
    """
    return file.tell() + len(message) > 1000000  # 1MB

logger.add("logs/app.log", rotation=should_rotate)

3. 日志过滤

from loguru import logger

def filter_function(record):
    """
    日志过滤函数
    :param record: 日志记录
    :return: 是否保留该日志
    """
    return "password" not in record["message"].lower()

logger.add("logs/filtered.log", filter=filter_function)

# 或者使用简单的过滤器
logger.add("logs/error_only.log", filter=lambda record: record["level"].name == "ERROR")

上下文管理

1. 绑定上下文信息

from loguru import logger

# 绑定用户信息
user_logger = logger.bind(user_id=123, ip="192.168.1.1")
user_logger.info("用户登录")

# 临时绑定
with logger.contextualize(request_id="abc-123"):
    logger.info("处理请求开始")
    logger.info("处理请求结束")

2. 结构化日志

from loguru import logger

# 使用额外字段
logger.info("用户操作", extra={
    "user_id": 123,
    "action": "login",
    "ip_address": "192.168.1.1",
    "timestamp": "2024-01-01T10:00:00"
})

# 自定义格式显示结构化数据
logger.add(
    "logs/structured.log",
    format="{time} | {level} | {extra[user_id]} | {message}",
    filter=lambda record: "user_id" in record["extra"]
)

性能优化

1. 延迟字符串格式化

from loguru import logger

# 推荐：延迟格式化
logger.info("用户 {} 执行了 {} 操作", user_id, action)

# 不推荐：立即格式化
logger.info(f"用户 {user_id} 执行了 {action} 操作")

2. 条件日志记录

from loguru import logger

# 仅在满足条件时记录
if logger.level("DEBUG").no >= logger.level("INFO").no:
    expensive_debug_info = get_expensive_debug_info()
    logger.debug("调试信息: {}", expensive_debug_info)

与其他库集成

1. 与标准库 logging 集成

import logging
from loguru import logger

class InterceptHandler(logging.Handler):
    """
    拦截标准库日志的处理器
    """
    
    def emit(self, record: logging.LogRecord) -> None:
        """
        处理日志记录
        :param record: 日志记录
        :return: None
        """
        # 获取对应的 loguru 级别
        try:
            level = logger.level(record.levelname).name
        except ValueError:
            level = record.levelno

        # 查找调用者
        frame, depth = logging.currentframe(), 2
        while frame.f_code.co_filename == logging.__file__:
            frame = frame.f_back
            depth += 1

        logger.opt(depth=depth, exception=record.exc_info).log(level, record.getMessage())

# 配置标准库日志
logging.basicConfig(handlers=[InterceptHandler()], level=0, force=True)

2. 与 FastAPI 集成

from fastapi import FastAPI
from loguru import logger
import sys

# 配置 loguru
logger.remove()
logger.add(sys.stdout, level="INFO")
logger.add("logs/api.log", rotation="1 day", level="DEBUG")

app = FastAPI()

@app.middleware("http")
async def log_requests(request, call_next):
    """
    记录HTTP请求的中间件
    :param request: HTTP请求
    :param call_next: 下一个处理函数
    :return: HTTP响应
    """
    start_time = time.time()
    
    logger.info("请求开始", extra={
        "method": request.method,
        "url": str(request.url),
        "client_ip": request.client.host
    })
    
    response = await call_next(request)
    
    process_time = time.time() - start_time
    logger.info("请求完成", extra={
        "status_code": response.status_code,
        "process_time": process_time
    })
    
    return response

最佳实践

1. 项目结构化日志配置

# config/logging_config.py
from loguru import logger
import sys
from pathlib import Path

def setup_logging(app_name: str, log_level: str = "INFO") -> None:
    """
    设置应用日志配置
    :param app_name: 应用名称
    :param log_level: 日志级别
    :return: None
    """
    # 创建日志目录
    log_dir = Path("logs")
    log_dir.mkdir(exist_ok=True)
    
    # 移除默认处理器
    logger.remove()
    
    # 控制台输出（彩色）
    logger.add(
        sys.stderr,
        format="<green>{time:YYYY-MM-DD HH:mm:ss}</green> | <level>{level: <8}</level> | <cyan>{name}</cyan>:<cyan>{function}</cyan>:<cyan>{line}</cyan> - <level>{message}</level>",
        level=log_level,
        colorize=True
    )
    
    # 一般日志文件
    logger.add(
        log_dir / f"{app_name}_{{time:YYYY-MM-DD}}.log",
        rotation="00:00",
        retention="30 days",
        compression="zip",
        level="DEBUG",
        format="{time:YYYY-MM-DD HH:mm:ss.SSS} | {level: <8} | {name}:{function}:{line} | {message}"
    )
    
    # 错误日志单独文件
    logger.add(
        log_dir / f"{app_name}_error_{{time:YYYY-MM-DD}}.log",
        rotation="00:00",
        retention="90 days",
        compression="zip",
        level="ERROR",
        format="{time:YYYY-MM-DD HH:mm:ss.SSS} | {level: <8} | {name}:{function}:{line} | {message} | {extra}"
    )

2. 装饰器工具

from loguru import logger
from functools import wraps
import time

def log_function_call(include_args: bool = True, include_result: bool = True):
    """
    记录函数调用的装饰器
    :param include_args: 是否包含参数
    :param include_result: 是否包含返回值
    :return: 装饰器函数
    """
    def decorator(func):
        @wraps(func)
        def wrapper(*args, **kwargs):
            start_time = time.time()
            
            # 记录函数开始
            if include_args:
                logger.debug(f"调用函数 {func.__name__}，参数: args={args}, kwargs={kwargs}")
            else:
                logger.debug(f"调用函数 {func.__name__}")
            
            try:
                result = func(*args, **kwargs)
                execution_time = time.time() - start_time
                
                # 记录成功结果
                if include_result:
                    logger.debug(f"函数 {func.__name__} 执行成功，耗时 {execution_time:.3f}s，返回值: {result}")
                else:
                    logger.debug(f"函数 {func.__name__} 执行成功，耗时 {execution_time:.3f}s")
                
                return result
                
            except Exception as e:
                execution_time = time.time() - start_time
                logger.error(f"函数 {func.__name__} 执行失败，耗时 {execution_time:.3f}s，错误: {e}")
                raise
                
        return wrapper
    return decorator

# 使用示例
@log_function_call(include_args=True, include_result=True)
def calculate_sum(a: int, b: int) -> int:
    """
    计算两数之和
    :param a: 第一个数
    :param b: 第二个数
    :return: 两数之和
    """
    return a + b

3. 异步日志记录

import asyncio
from loguru import logger

async def async_log_processor():
    """
    异步日志处理器
    :return: None
    """
    while True:
        # 模拟异步日志处理
        await asyncio.sleep(1)
        logger.info("异步日志处理中...")

# 在异步环境中使用
async def main():
    """
    主异步函数
    :return: None
    """
    # 启动异步日志处理任务
    log_task = asyncio.create_task(async_log_processor())
    
    # 其他异步操作
    await asyncio.sleep(5)
    
    # 取消日志任务
    log_task.cancel()

# 运行示例
# asyncio.run(main())

常见问题和解决方案

1. 日志文件权限问题

from loguru import logger
import os

def safe_add_file_handler(file_path: str, **kwargs) -> None:
    """
    安全地添加文件处理器
    :param file_path: 文件路径
    :param kwargs: 其他参数
    :return: None
    """
    try:
        # 确保目录存在
        os.makedirs(os.path.dirname(file_path), exist_ok=True)
        logger.add(file_path, **kwargs)
    except PermissionError:
        logger.warning(f"无法写入日志文件 {file_path}，使用控制台输出")
        logger.add(sys.stderr, **kwargs)

2. 内存使用优化

from loguru import logger

# 限制日志缓冲区大小
logger.add(
    "logs/app.log",
    buffering=1,  # 减少缓冲
    enqueue=True,  # 使用队列避免阻塞
)

# 定期清理处理器
def cleanup_old_handlers():
    """
    清理旧的日志处理器
    :return: None
    """
    logger.remove()
    # 重新添加必要的处理器
    setup_logging()

配置模板

开发环境配置

from loguru import logger
import sys

def setup_dev_logging():
    """
    开发环境日志配置
    :return: None
    """
    logger.remove()
    logger.add(
        sys.stderr,
        format="<green>{time:HH:mm:ss}</green> | <level>{level: <8}</level> | <cyan>{name}</cyan>:<cyan>{function}</cyan> - <level>{message}</level>",
        level="DEBUG",
        colorize=True
    )

生产环境配置

from loguru import logger

def setup_prod_logging():
    """
    生产环境日志配置
    :return: None
    """
    logger.remove()
    
    # 结构化JSON日志
    logger.add(
        "logs/app.log",
        format="{time:YYYY-MM-DD HH:mm:ss.SSS} | {level} | {name}:{function}:{line} | {message}",
        level="INFO",
        rotation="100 MB",
        retention="30 days",
        compression="gz",
        serialize=True  # JSON格式
    )
    
    # 错误日志
    logger.add(
        "logs/error.log",
        level="ERROR",
        rotation="10 MB",
        retention="90 days"
    )

总结

Loguru 提供了简洁而强大的日志功能，主要优势包括：

1. 开箱即用 - 无需复杂配置

2. 自动异常捕获 - 使用 @logger.catch 装饰器

3. 灵活的格式化 - 支持彩色输出和自定义格式

4. 智能轮转 - 按时间、大小或自定义条件轮转

5. 结构化日志 - 支持额外字段和上下文信息

6. 性能优化 - 延迟格式化和异步处理

通过合理配置和使用最佳实践，Loguru 可以满足从简单脚本到复杂应用的各种日志需求。