2853 字
14 分钟

Python 异常处理与日志系统完全指南：从 try-except 到结构化日志实战

2026-07-23
Python教程
Python
/
异常处理
/
日志系统
/
logging
/
try-except
/
错误处理

异常处理和日志记录是编写健壮 Python 程序的基础技能。良好的错误处理让程序优雅降级，完善的日志系统让问题排查有迹可循。本文从基础语法到生产实践，全面讲解 Python 异常处理与日志系统。

Python 异常处理与日志系统

本文内容包括：

  • try-except-else-finally 完整语法
  • 异常类型与继承体系
  • 自定义异常类设计
  • raise 与异常链
  • logging 模块深度配置
  • Handler、Formatter、Filter 详解
  • Loguru 现代日志库
  • 结构化日志与 JSON 输出
  • 生产环境日志最佳实践
  • 异常处理设计模式

一、异常处理基础#

1.1 try-except 基本语法#

# 基础异常捕获
try:
    result = 10 / 0
except ZeroDivisionError:
    print("不能除以零")


# 捕获多个异常
try:
    with open("data.txt", "r") as f:
        data = json.load(f)
except FileNotFoundError:
    print("文件不存在")
except json.JSONDecodeError:
    print("JSON 格式错误")


# 捕获多个异常（统一处理）
try:
    risky_operation()
except (ValueError, TypeError) as e:
    print(f"参数错误: {e}")

1.2 try-except-else-finally#

def read_config(path):
    try:
        # 尝试执行
        with open(path, "r") as f:
            content = f.read()
        config = json.loads(content)
    except FileNotFoundError:
        # 异常时执行
        print(f"配置文件 {path} 不存在")
        return {}
    except json.JSONDecodeError as e:
        # 异常时执行
        print(f"配置格式错误: {e}")
        return {}
    else:
        # 无异常时执行
        print("配置加载成功")
        return config
    finally:
        # 无论是否异常都执行
        print("读取操作完成")


# 执行顺序测试
def test_order():
    try:
        print("1. try")
        raise ValueError("测试异常")
    except ValueError:
        print("2. except")
    else:
        print("3. else (不会执行)")
    finally:
        print("4. finally")


test_order()
# 输出：
# 1. try
# 2. except
# 4. finally

1.3 捕获所有异常#

# 方法1：捕获 Exception（推荐）
try:
    risky_operation()
except Exception as e:
    print(f"发生错误: {type(e).__name__}: {e}")


# 方法2：捕获 BaseException（包含系统退出信号，慎用）
try:
    risky_operation()
except BaseException as e:
    print(f"发生错误: {e}")


# 方法3：空 except（不推荐，捕获所有包括 KeyboardInterrupt）
try:
    risky_operation()
except:
    print("发生错误")

1.4 异常信息获取#

try:
    1 / 0
except ZeroDivisionError as e:
    # 异常对象
    print(f"异常类型: {type(e).__name__}")   # ZeroDivisionError
    print(f"异常信息: {e}")                   # division by zero


    # 完整堆栈
    import traceback
    print("堆栈追踪:")
    traceback.print_exc()


    # 格式化为字符串
    stack_trace = traceback.format_exc()
    print(f"完整追踪:\n{stack_trace}")

二、异常类型与继承体系#

2.1 Python 内置异常层次#

BaseException
├── SystemExit              # sys.exit() 触发
├── KeyboardInterrupt       # Ctrl+C 触发
├── GeneratorExit           # 生成器关闭时
└── Exception               # 所有常规异常的基类
    ├── ArithmeticError
    │   └── ZeroDivisionError
    ├── LookupError
    │   ├── IndexError
    │   └── KeyError
    ├── TypeError
    ├── ValueError
    │   └── UnicodeDecodeError
    ├── RuntimeError
    │   └── RecursionError
    ├── OSError
    │   ├── FileNotFoundError
    │   ├── PermissionError
    │   └── TimeoutError
    ├── ImportError
    │   └── ModuleNotFoundError
    ├── AttributeError
    ├── NameError
    │   └── UnboundLocalError
    ├── AssertionError
    ├── StopIteration
    └── Warning
        ├── UserWarning
        ├── DeprecationWarning
        └── RuntimeWarning

2.2 常见异常场景#

# ValueError：值不符合要求
try:
    int("abc")
except ValueError as e:
    print(f"无法转换: {e}")


# TypeError：类型错误
try:
    len(123)
except TypeError as e:
    print(f"类型错误: {e}")


# KeyError：字典键不存在
data = {"name": "Alice"}
try:
    age = data["age"]
except KeyError:
    age = data.get("age", 0)  # 安全获取


# IndexError：索引越界
items = [1, 2, 3]
try:
    item = items[10]
except IndexError:
    item = None


# AttributeError：属性不存在
try:
    obj.nonexistent_method()
except AttributeError:
    print("方法不存在")

三、自定义异常类#

3.1 基础自定义异常#

# 基础异常类
class BusinessError(Exception):
    """业务逻辑异常基类"""
    pass


class ValidationError(BusinessError):
    """数据验证失败"""
    pass


class NotFoundError(BusinessError):
    """资源不存在"""
    pass


class PermissionDeniedError(BusinessError):
    """权限不足"""
    pass


# 使用
class UserService:
    def get_user(self, user_id: int):
        if user_id <= 0:
            raise ValidationError("用户ID必须为正整数")


        user = self.db.query(user_id)
        if not user:
            raise NotFoundError(f"用户 {user_id} 不存在")


        return user


# 捕获
try:
    user = service.get_user(-1)
except ValidationError as e:
    print(f"参数错误: {e}")
except NotFoundError as e:
    print(f"未找到: {e}")
except BusinessError as e:
    print(f"业务错误: {e}")

3.2 带状态码的异常#

class APIError(Exception):
    """API 异常，包含状态码和详情"""


    def __init__(self, message: str, status_code: int = 500, details: dict = None):
        super().__init__(message)
        self.status_code = status_code
        self.details = details or {}


    def to_dict(self) -> dict:
        return {
            "error": str(self),
            "status_code": self.status_code,
            "details": self.details
        }


class BadRequestError(APIError):
    def __init__(self, message: str, details: dict = None):
        super().__init__(message, 400, details)


class UnauthorizedError(APIError):
    def __init__(self, message: str = "未授权"):
        super().__init__(message, 401)


class ForbiddenError(APIError):
    def __init__(self, message: str = "禁止访问"):
        super().__init__(message, 403)


class NotFoundError(APIError):
    def __init__(self, resource: str):
        super().__init__(f"{resource} 不存在", 404)


# 使用
def get_user(user_id: int):
    if not current_user.is_authenticated:
        raise UnauthorizedError()


    user = db.query(user_id)
    if not user:
        raise NotFoundError("用户")


    return user


# 异常处理装饰器
def handle_api_errors(func):
    def wrapper(*args, **kwargs):
        try:
            return func(*args, **kwargs)
        except APIError as e:
            return jsonify(e.to_dict()), e.status_code
        except Exception as e:
            logger.exception("未预期的错误")
            return jsonify({"error": "服务器内部错误"}), 500
    return wrapper

3.3 异常链（Exception Chaining）#

# 保留原始异常信息
def parse_config(path: str):
    try:
        with open(path, "r") as f:
            content = f.read()
        return json.loads(content)
    except FileNotFoundError as e:
        raise ConfigError(f"配置文件缺失: {path}") from e
    except json.JSONDecodeError as e:
        raise ConfigError(f"配置文件格式错误: {path}") from e


# 使用
try:
    config = parse_config("app.json")
except ConfigError as e:
    print(f"配置错误: {e}")
    print(f"原始错误: {e.__cause__}")  # 查看原始异常


# 显式链式异常
try:
    risky_operation()
except ValueError:
    raise BusinessError("业务处理失败") from None  # 不保留原始异常

四、上下文管理器与异常#

4.1 自定义上下文管理器#

from contextlib import contextmanager


@contextmanager
def database_connection():
    """数据库连接上下文管理器"""
    conn = create_connection()
    try:
        yield conn
    except Exception:
        conn.rollback()
        raise
    finally:
        conn.close()


# 使用
with database_connection() as conn:
    conn.execute("INSERT INTO users ...")
    conn.commit()


# 带重试的上下文管理器
import time
from functools import wraps


@contextmanager
def retry_on_error(max_retries=3, delay=1, exceptions=(Exception,)):
    """出错时自动重试"""
    for attempt in range(max_retries):
        try:
            yield attempt
            return
        except exceptions as e:
            if attempt == max_retries - 1:
                raise
            print(f"第 {attempt + 1} 次失败: {e}，{delay}秒后重试...")
            time.sleep(delay)


# 使用
with retry_on_error(max_retries=3, exceptions=(ConnectionError,)):
    fetch_data_from_api()

4.2 suppress 上下文管理器#

from contextlib import suppress


# 忽略特定异常
with suppress(FileNotFoundError):
    os.remove("temp_file.txt")


# 等价于
try:
    os.remove("temp_file.txt")
except FileNotFoundError:
    pass

五、logging 模块深度配置#

5.1 基础配置#

import logging


# 基础配置
logging.basicConfig(
    level=logging.INFO,
    format="%(asctime)s - %(name)s - %(levelname)s - %(message)s",
    datefmt="%Y-%m-%d %H:%M:%S"
)


logger = logging.getLogger("myapp")
logger.info("应用启动")
logger.warning("配置项缺失，使用默认值")
logger.error("数据库连接失败")

5.2 日志级别#

级别数值用途
DEBUG10调试信息，开发时使用
INFO20正常运行信息
WARNING30警告，程序仍正常运行
ERROR40错误，某项功能失败
CRITICAL50严重错误，程序可能崩溃
# 设置不同级别的日志
logger.debug("调试信息: %s", debug_data)
logger.info("用户 %s 登录", username)
logger.warning("磁盘空间不足: %d%%", disk_usage)
logger.error("请求失败: %s", error_msg)
logger.critical("系统内存耗尽，即将退出")

5.3 高级配置（代码方式）#

import logging
import logging.handlers
import sys


def setup_logging(name: str = "app", level: int = logging.INFO):
    """配置日志系统"""


    logger = logging.getLogger(name)
    logger.setLevel(level)


    # 清除已有处理器
    logger.handlers = []


    # 格式
    formatter = logging.Formatter(
        "%(asctime)s | %(name)s | %(levelname)-8s | %(filename)s:%(lineno)d | %(message)s",
        datefmt="%Y-%m-%d %H:%M:%S"
    )


    # 1. 控制台输出
    console_handler = logging.StreamHandler(sys.stdout)
    console_handler.setLevel(logging.INFO)
    console_handler.setFormatter(formatter)
    logger.addHandler(console_handler)


    # 2. 文件输出（按大小轮转）
    file_handler = logging.handlers.RotatingFileHandler(
        "app.log",
        maxBytes=10 * 1024 * 1024,  # 10MB
        backupCount=5
    )
    file_handler.setLevel(logging.DEBUG)
    file_handler.setFormatter(formatter)
    logger.addHandler(file_handler)


    # 3. 错误日志（单独文件）
    error_handler = logging.handlers.RotatingFileHandler(
        "error.log",
        maxBytes=10 * 1024 * 1024,
        backupCount=10
    )
    error_handler.setLevel(logging.ERROR)
    error_handler.setFormatter(formatter)
    logger.addHandler(error_handler)


    # 4. 按时间轮转（每天一个文件）
    daily_handler = logging.handlers.TimedRotatingFileHandler(
        "daily.log",
        when="midnight",
        interval=1,
        backupCount=30
    )
    daily_handler.setLevel(logging.INFO)
    daily_handler.setFormatter(formatter)
    logger.addHandler(daily_handler)


    return logger


# 使用
logger = setup_logging()
logger.info("应用启动")

5.4 配置文件方式#

logging.yaml
version: 1
disable_existing_loggers: false


formatters:
  standard:
    format: "%(asctime)s | %(name)s | %(levelname)-8s | %(message)s"
    datefmt: "%Y-%m-%d %H:%M:%S"
  detailed:
    format: "%(asctime)s | %(name)s | %(levelname)-8s | %(filename)s:%(lineno)d | %(funcName)s | %(message)s"
    datefmt: "%Y-%m-%d %H:%M:%S"


handlers:
  console:
    class: logging.StreamHandler
    level: INFO
    formatter: standard
    stream: ext://sys.stdout


  file:
    class: logging.handlers.RotatingFileHandler
    level: DEBUG
    formatter: detailed
    filename: logs/app.log
    maxBytes: 10485760  # 10MB
    backupCount: 5


  error_file:
    class: logging.handlers.RotatingFileHandler
    level: ERROR
    formatter: detailed
    filename: logs/error.log
    maxBytes: 10485760
    backupCount: 10


loggers:
  myapp:
    level: DEBUG
    handlers: [console, file, error_file]
    propagate: false


root:
  level: WARNING
  handlers: [console]
import logging.config
import yaml


# 从 YAML 加载配置
with open("logging.yaml", "r") as f:
    config = yaml.safe_load(f)
    logging.config.dictConfig(config)


logger = logging.getLogger("myapp")

六、Loguru 现代日志库#

6.1 Loguru 基础用法#

# 安装: pip install loguru
from loguru import logger


# 开箱即用，无需配置
logger.info("这是一条信息")
logger.warning("这是一条警告")
logger.error("这是一条错误")


# 格式化输出
logger.info("用户 {} 执行了 {}", "Alice", "delete")


# 结构化日志
logger.info("用户登录", user_id=123, ip="192.168.1.1")

6.2 Loguru 高级配置#

from loguru import logger
import sys


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


# 添加控制台输出
logger.add(
    sys.stdout,
    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="INFO",
    colorize=True
)


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


# 添加错误日志（单独文件）
logger.add(
    "logs/error.log",
    rotation="10 MB",
    retention="10 days",
    level="ERROR",
    backtrace=True,       # 详细回溯
    diagnose=True         # 变量诊断
)


# 使用
logger.info("应用启动")
logger.error("发生错误", exc_info=True)

6.3 Loguru 异常捕获#

from loguru import logger


# 自动捕获异常并记录
@logger.catch
def risky_function(x, y):
    return x / y


# 调用时出错会自动记录
risky_function(1, 0)  # 自动记录完整异常信息


# 自定义异常处理
@logger.catch(reraise=True)
def must_raise():
    raise ValueError("必须抛出的异常")


# 上下文管理器
with logger.catch():
    1 / 0  # 自动记录


# 异步异常捕获
@logger.catch
async def async_risky():
    raise RuntimeError("异步错误")

6.4 结构化日志与上下文#

from loguru import logger


# 添加上下文绑定
logger = logger.bind(request_id="req-123", user="Alice")
logger.info("处理请求")  # 自动包含 request_id 和 user


# 上下文管理器
with logger.contextualize(task_id="task-456"):
    logger.info("执行任务")  # 包含 task_id
    logger.info("任务完成")


# JSON 结构化输出（生产环境推荐）
logger.remove()
logger.add("logs/app.json", serialize=True)


logger.info("用户登录", user_id=123, ip="192.168.1.1")
# 输出:
# {
#   "text": "用户登录",
#   "record": {
#     "time": {"repr": "2026-07-23 10:00:00", "timestamp": 1753236000},
#     "level": {"name": "INFO", "no": 20},
#     "extra": {"user_id": 123, "ip": "192.168.1.1"}
#   }
# }

七、生产环境最佳实践#

7.1 异常处理模式#

# 模式1：提前返回（Guard Clause）
def process_user(user_id: int) -> dict:
    if user_id <= 0:
        raise ValueError("用户ID必须为正整数")


    user = db.get_user(user_id)
    if not user:
        raise NotFoundError(f"用户 {user_id} 不存在")


    if not user.is_active:
        raise ForbiddenError("用户已被禁用")


    return user.to_dict()


# 模式2：异常转换
def fetch_data():
    try:
        response = requests.get(API_URL, timeout=5)
        response.raise_for_status()
        return response.json()
    except requests.Timeout:
        raise APIError("请求超时", status_code=504)
    except requests.HTTPError as e:
        raise APIError(f"HTTP错误: {e.response.status_code}", status_code=e.response.status_code)


# 模式3：回退策略
def get_data_with_fallback():
    try:
        return fetch_from_primary()
    except PrimaryError:
        logger.warning("主数据源失败，使用备用数据源")
        return fetch_from_backup()

7.2 日志记录规范#

import logging
from contextvars import ContextVar


# 请求ID上下文
request_id_var: ContextVar[str] = ContextVar("request_id", default="")


class RequestIdFilter(logging.Filter):
    """自动添加请求ID到日志"""
    def filter(self, record):
        record.request_id = request_id_var.get()
        return True


# 配置
logger = logging.getLogger("app")
handler = logging.StreamHandler()
handler.setFormatter(logging.Formatter(
    "%(asctime)s | %(request_id)s | %(levelname)s | %(message)s"
))
handler.addFilter(RequestIdFilter())
logger.addHandler(handler)


# 使用
from contextlib import contextmanager


@contextmanager
def set_request_id(request_id: str):
    token = request_id_var.set(request_id)
    try:
        yield
    finally:
        request_id_var.reset(token)


# 在请求处理中
with set_request_id("req-123"):
    logger.info("处理请求")
    # 输出: 2026-07-23 10:00:00 | req-123 | INFO | 处理请求

7.3 性能优化#

# 避免字符串拼接开销
# ❌ 不好（每次都会执行字符串拼接）
logger.debug("Data: " + expensive_operation())


# ✅ 好（使用延迟求值）
logger.debug("Data: {}", expensive_operation())


# ✅ 更好（先检查级别）
if logger.isEnabledFor(logging.DEBUG):
    logger.debug("Data: %s", expensive_operation())


# Loguru 自动优化（使用 {} 格式化）
logger.debug("Data: {}", expensive_operation())  # 自动延迟求值

八、总结#

异常处理检查清单#

✅ 捕获具体异常，不要用裸 except
✅ 使用 try-except-else-finally 完整结构
✅ 自定义异常继承合适的基类
✅ 使用 raise ... from ... 保留异常链
✅ 异常信息要包含上下文
✅ 不要吞掉异常（至少记录日志）
✅ 资源清理用 finally 或上下文管理器

日志系统检查清单#

✅ 使用结构化日志（JSON）便于解析
✅ 添加请求ID便于追踪
✅ 不同级别输出到不同目的地
✅ 配置日志轮转防止磁盘占满
✅ 敏感信息（密码、token）不要记录
✅ 异常使用 logger.exception() 自动记录堆栈
✅ 生产环境考虑异步日志避免阻塞

logging vs Loguru 对比#

特性loggingLoguru
配置复杂度复杂简单
结构化日志需手动实现原生支持
异常捕获手动@logger.catch
异步支持需额外配置内置
日志轮转需 Handler内置
依赖标准库第三方

建议：新项目优先使用 Loguru，legacy 项目逐步迁移。核心原则是：良好的异常处理让程序优雅降级，完善的日志让问题有迹可循。

Python 异常处理与日志系统完全指南：从 try-except 到结构化日志实战
https://971918.xyz/posts/python-guide/python-exception-logging-guide/
作者
九所长
发布于
2026-07-23
许可协议
CC BY-NC-SA 4.0
Nginx 反向代理与负载均衡完全指南：从基础配置到高可用架构实战
Python 项目工程化实战（三）：mypy 类型检查 + uv build 打包发布 PyPI 全流程 | Python 进阶
© 2026 九所长. All Rights Reserved. / 关于我们 / 友情链接 / 免责声明 / 服务条款 / 隐私政策 / RSS / Sitemap
Powered by Astro & Fuwari