Python日志调试案例如何实现？

wen python案例 2026-06-06 18:47:34 3

本文目录导读：

基础日志配置
实用调试案例 - 函数调用追踪
性能日志调试
网络请求调试
数据库操作调试
高级：日志级别动态控制
完整调试工具类
使用建议

我来介绍几种Python日志调试的实现方案,从简单到复杂：

基础日志配置

import logging
import sys
# 基本的日志配置
logging.basicConfig(
    level=logging.DEBUG,
    format='%(asctime)s - %(name)s - %(levelname)s - %(message)s',
    handlers=[
        logging.StreamHandler(sys.stdout),  # 输出到控制台
        logging.FileHandler('app.log')      # 输出到文件
    ]
)
logger = logging.getLogger(__name__)
# 示例函数
def divide_numbers(a, b):
    logger.debug(f"尝试计算 {a} / {b}")
    try:
        result = a / b
        logger.info(f"计算成功: {result}")
        return result
    except ZeroDivisionError as e:
        logger.error(f"除零错误: {e}")
        return None
    except Exception as e:
        logger.exception(f"未知错误: {e}")
        return None
# 测试
divide_numbers(10, 2)  # 正常
divide_numbers(10, 0)  # 除零错误

实用调试案例 - 函数调用追踪

import logging
from functools import wraps
logging.basicConfig(level=logging.DEBUG, format='%(levelname)s - %(message)s')
logger = logging.getLogger(__name__)
def log_function_call(func):
    """装饰器：记录函数调用和返回值"""
    @wraps(func)
    def wrapper(*args, **kwargs):
        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}")
            raise
    return wrapper
@log_function_call
def calculate_price(price, tax_rate):
    return price * (1 + tax_rate)
# 测试
result = calculate_price(100, 0.1)
print(f"最终价格: {result}")
# 错误测试
try:
    calculate_price("100", 0.1)  # 类型错误
except Exception as e:
    print(f"捕获错误: {e}")

性能日志调试

import logging
import time
import random
logging.basicConfig(level=logging.DEBUG, format='%(asctime)s - %(levelname)s - %(message)s')
logger = logging.getLogger(__name__)
class PerformanceLogger:
    def __init__(self, threshold=0.5):
        self.threshold = threshold  # 性能阈值（秒）
    def __enter__(self):
        self.start_time = time.time()
        return self
    def __exit__(self, exc_type, exc_val, exc_tb):
        elapsed = time.time() - self.start_time
        if elapsed > self.threshold:
            logger.warning(f"操作执行时间 {elapsed:.2f}s 超过阈值 {self.threshold}s")
        else:
            logger.info(f"操作执行时间 {elapsed:.2f}s")
        return False  # 不抑制异常
def process_data(data_size):
    with PerformanceLogger(threshold=0.2):
        logger.debug(f"开始处理 {data_size} 条数据")
        total = 0
        for i in range(data_size):
            total += random.randint(1, 100)
            if i % 1000 == 0:
                logger.debug(f"已处理 {i} 条数据")
        logger.info(f"处理完成，总和: {total}")
        return total
# 测试
process_data(1000)  # 测试小数据量
process_data(100000)  # 测试大数据量（可能超阈值）

网络请求调试

import logging
import requests
import json
logging.basicConfig(level=logging.DEBUG, format='%(levelname)s - %(message)s')
logger = logging.getLogger(__name__)
def fetch_api_data(url, timeout=10):
    """带详细日志的API请求函数"""
    logger.info(f"开始请求: {url}")
    try:
        # 记录请求信息
        logger.debug(f"请求参数 - timeout: {timeout}s")
        # 发送请求
        response = requests.get(url, timeout=timeout)
        # 记录响应状态
        logger.debug(f"响应状态码: {response.status_code}")
        logger.debug(f"响应头: {dict(response.headers)}")
        # 检查状态码
        if response.status_code == 200:
            data = response.json()
            logger.info(f"请求成功，数据大小: {len(str(data))} 字符")
            return data
        elif response.status_code == 404:
            logger.warning(f"资源未找到: {url}")
            return None
        else:
            logger.error(f"请求失败，状态码: {response.status_code}")
            return None
    except requests.Timeout:
        logger.error(f"请求超时 (timeout={timeout}s): {url}")
        return None
    except requests.ConnectionError as e:
        logger.error(f"连接错误: {e}")
        return None
    except json.JSONDecodeError as e:
        logger.error(f"JSON解析错误: {e}")
        return None
    except Exception as e:
        logger.exception(f"未预期的错误: {e}")
        return None
# 测试
result = fetch_api_data("https://api.github.com/users/octocat")
if result:
    logger.info(f"用户: {result.get('login')}")
# 测试错误情况
result = fetch_api_data("https://api.github.com/users/nonexistent_user_12345")

数据库操作调试

import logging
import sqlite3
from contextlib import contextmanager
logging.basicConfig(level=logging.DEBUG, format='%(levelname)s - %(message)s')
logger = logging.getLogger(__name__)
@contextmanager
def database_logger(db_path):
    """上下文管理器：自动记录数据库操作"""
    conn = sqlite3.connect(db_path)
    logger.info(f"连接到数据库: {db_path}")
    try:
        yield conn
    except sqlite3.Error as e:
        logger.error(f"数据库错误: {e}")
        conn.rollback()
        raise
    else:
        conn.commit()
        logger.info("数据库事务已提交")
    finally:
        conn.close()
        logger.info("数据库连接已关闭")
def execute_query_with_log(conn, query, params=None):
    """执行SQL查询并记录日志"""
    logger.debug(f"执行SQL: {query}")
    logger.debug(f"参数: {params}")
    cursor = conn.cursor()
    try:
        if params:
            cursor.execute(query, params)
        else:
            cursor.execute(query)
        rows = cursor.fetchall()
        logger.info(f"查询返回 {len(rows)} 行数据")
        return rows
    except sqlite3.Error as e:
        logger.error(f"SQL错误: {e}")
        raise
# 使用示例
def database_debug_example():
    with database_logger(":memory:") as conn:
        # 创建表
        create_table("CREATE TABLE users (id INTEGER PRIMARY KEY, name TEXT, age INTEGER)")
        # 插入数据
        insert_data = [
            ("Alice", 30),
            ("Bob", 25),
            ("Charlie", 35)
        ]
        for name, age in insert_data:
            execute_query_with_log(
                conn, 
                "INSERT INTO users (name, age) VALUES (?, ?)",
                (name, age)
            )
        # 查询数据
        result = execute_query_with_log(conn, "SELECT * FROM users WHERE age > ?", (28,))
        logger.info(f"年龄大于28的用户: {result}")
# 运行测试
database_debug_example()

高级：日志级别动态控制

import logging
import sys
class DynamicLogger:
    """可动态调整日志级别的日志管理器"""
    def __init__(self, name="DynamicLogger"):
        self.logger = logging.getLogger(name)
        self.logger.setLevel(logging.DEBUG)
        # 移除已有的处理器
        self.logger.handlers.clear()
        # 添加控制台处理器
        console_handler = logging.StreamHandler(sys.stdout)
        console_handler.setLevel(logging.DEBUG)
        # 设置格式
        formatter = logging.Formatter(
            '%(asctime)s - %(name)s - %(levelname)s - %(message)s'
        )
        console_handler.setFormatter(formatter)
        self.logger.addHandler(console_handler)
        self.level_map = {
            'DEBUG': logging.DEBUG,
            'INFO': logging.INFO,
            'WARNING': logging.WARNING,
            'ERROR': logging.ERROR,
            'CRITICAL': logging.CRITICAL
        }
    def set_level(self, level_name):
        if level_name in self.level_map:
            self.logger.setLevel(self.level_map[level_name])
            self.logger.info(f"日志级别已更改为: {level_name}")
        else:
            self.logger.error(f"无效的日志级别: {level_name}")
    def debug(self, message):
        self.logger.debug(message)
    def info(self, message):
        self.logger.info(message)
    def warning(self, message):
        self.logger.warning(message)
    def error(self, message):
        self.logger.error(message)
# 使用示例
def dynamic_logger_example():
    dlog = DynamicLogger()
    dlog.info("系统初始化中...")
    # 初始时只有 INFO 及以上级别会显示
    dlog.set_level('DEBUG')
    dlog.debug("这是个调试信息")
    dlog.info("信息显示正常")
    # 改变级别
    dlog.set_level('WARNING')
    dlog.info("这行不会显示")  # 低于 WARNING 级别
    dlog.warning("警告信息")
    dlog.error("错误信息")
dynamic_logger_example()

完整调试工具类

import logging
import traceback
from datetime import datetime
class DebugTool:
    """综合调试工具类"""
    def __init__(self, log_file=None):
        self.setup_logging(log_file)
        self.logger = logging.getLogger(__name__)
    def setup_logging(self, log_file=None):
        """设置日志配置"""
        handlers = [logging.StreamHandler()]
        if log_file:
            handlers.append(logging.FileHandler(log_file))
        logging.basicConfig(
            level=logging.DEBUG,
            format='%(asctime)s - %(levelname)s - %(message)s',
            handlers=handlers
        )
    def debug_variable(self, name, value):
        """调试变量"""
        self.logger.debug(f"变量 {name} = {repr(value)} (类型: {type(value).__name__})")
    def debug_block(self, block_name, start=True):
        """标记代码块开始/结束"""
        if start:
            self.logger.debug(f"====== 进入代码块: {block_name} ======")
        else:
            self.logger.debug(f"====== 离开代码块: {block_name} ======")
    def catch_exception(self, func):
        """异常捕获装饰器"""
        def wrapper(*args, **kwargs):
            try:
                return func(*args, **kwargs)
            except Exception as e:
                self.logger.error(f"函数 {func.__name__} 发生错误:")
                self.logger.error(f"  错误类型: {type(e).__name__}")
                self.logger.error(f"  错误信息: {str(e)}")
                self.logger.error(f"  调用栈:\n{traceback.format_exc()}")
                raise
        return wrapper
    def time_execution(self, func):
        """执行时间测量装饰器"""
        def wrapper(*args, **kwargs):
            start = datetime.now()
            result = func(*args, **kwargs)
            end = datetime.now()
            duration = (end - start).total_seconds()
            self.logger.info(f"函数 {func.__name__} 执行时间: {duration:.3f}秒")
            return result
        return wrapper
# 使用示例
debug_tool = DebugTool(log_file="debug.log")
class Calculator:
    def __init__(self):
        self.logger = debug_tool.logger
    @debug_tool.catch_exception
    @debug_tool.time_execution
    def complex_calculation(self, x, y, z):
        debug_tool.debug_block("复杂计算")
        debug_tool.debug_variable("x", x)
        debug_tool.debug_variable("y", y)
        debug_tool.debug_variable("z", z)
        # 一些复杂的计算
        result = (x + y) * z / (x - y)
        debug_tool.debug_variable("result", result)
        debug_tool.debug_block("复杂计算", start=False)
        return result
# 测试
calc = Calculator()
try:
    result = calc.complex_calculation(10, 5, 3)
    print(f"计算结果: {result}")
except Exception:
    print("计算出错，详情请查看日志")

使用建议

选择合适的日志级别：
- DEBUG：详细调试信息
- INFO：正常操作信息
- WARNING：潜在问题
- ERROR：错误信息
- CRITICAL：严重错误

避免在生产环境使用 DEBUG 级别，使用配置文件控制：

# config.py
LOG_LEVEL = 'DEBUG'  # 开发环境
# LOG_LEVEL = 'INFO'  # 生产环境

注意敏感信息，不要在日志中记录密码、API密钥等。

这些案例覆盖了从基础到高级的各种调试场景,可以根据实际需求选择使用。

标签： Python案例

本文地址： https://dfhcn.com/post/1047.html

文章来源： wen