BLOG

用心记录思想

0%

Python Exception Handling Mechanisms and Design Principles Explained

Posted on
Symbols count in article: 2.7k Reading time ≈ 10 mins.

Python异常处理设计原则详解

1. 只在适当的层级处理异常

核心思想

异常应该在能够真正处理问题的层级被捕获，而不是在问题发生的地方盲目处理。

错误示范

1
2
3
4
5
6
7
8
9
10
11
12
13
# ❌ 在低层级过度处理异常
def read_config_file():
    try:
        with open("config.json", "r") as f:
            return json.load(f)
    except FileNotFoundError:
        return {}  # 静默返回空字典，掩盖了问题
    except json.JSONDecodeError:
        return {}  # 静默返回空字典，掩盖了问题

def main():
    config = read_config_file()
    # 后续代码可能因为配置缺失而失败，但不知道根本原因

正确做法

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
# ✅ 在适当的层级处理异常
def read_config_file():
    """读取配置文件，如果出现问题直接抛出异常"""
    try:
        with open("config.json", "r") as f:
            return json.load(f)
    except FileNotFoundError as e:
        raise ConfigurationError("配置文件不存在") from e
    except json.JSONDecodeError as e:
        raise ConfigurationError("配置文件格式错误") from e

def load_default_config():
    """提供默认配置"""
    return {"host": "localhost", "port": 8080}

def main():
    try:
        config = read_config_file()
    except ConfigurationError:
        # 在应用层级决定如何处理：使用默认配置并记录日志
        logger.warning("使用默认配置，配置文件加载失败")
        config = load_default_config()
    
    start_server(config)

分层处理策略

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
class DataAccessLayer:
    """数据访问层 - 只负责技术异常"""
    def get_user(self, user_id):
        try:
            return self.db.execute("SELECT * FROM users WHERE id = ?", user_id)
        except DatabaseConnectionError as e:
            # 技术异常：直接向上抛
            raise
        except sqlite3.OperationalError as e:
            # 技术异常：直接向上抛
            raise DataAccessError(f"数据库操作失败: {e}") from e

class BusinessLogicLayer:
    """业务逻辑层 - 处理业务异常"""
    def get_user_profile(self, user_id):
        try:
            user = self.data_access.get_user(user_id)
            if not user:
                # 业务规则：用户不存在是业务异常
                raise UserNotFoundError(f"用户 {user_id} 不存在")
            return UserProfile(user)
        except DataAccessError as e:
            # 技术异常转换为业务异常
            raise BusinessOperationError("无法获取用户数据") from e

class PresentationLayer:
    """表现层 - 最终异常处理"""
    def show_user_profile(self, user_id):
        try:
            profile = self.business_logic.get_user_profile(user_id)
            return self.render_template("profile.html", profile=profile)
        except UserNotFoundError:
            return self.render_error_page("用户不存在")
        except BusinessOperationError:
            return self.render_error_page("系统繁忙，请稍后重试")
        except Exception as e:
            logger.exception("未预期的错误")
            return self.render_error_page("系统内部错误")

2. 避免过度捕获异常（不要捕获所有异常）

核心思想

只捕获你知道如何处理的异常类型，让其他异常继续传播。

错误示范

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
# ❌ 过度捕获异常
def process_data(data):
    try:
        result = complex_data_processing(data)
        return result
    except:
        # 捕获所有异常，包括KeyboardInterrupt、SystemExit等
        return None  # 静默失败，难以调试

# ❌ 捕获过于宽泛的异常
def save_to_database(record):
    try:
        self.db.save(record)
    except Exception as e:  # 过于宽泛
        logger.error("保存失败")
        # 不知道具体是什么错误，可能是连接问题、权限问题、数据问题等

正确做法

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
# ✅ 精确捕获异常
def process_data(data):
    try:
        result = complex_data_processing(data)
        return result
    except ValueError as e:
        # 只处理数据格式错误
        logger.warning(f"数据格式错误: {e}")
        return sanitize_data(data)
    except ProcessingTimeout as e:
        # 只处理超时错误
        logger.warning("处理超时，重试中...")
        return self.retry_processing(data)
    # 其他异常继续传播

# ✅ 分层捕获策略
def save_to_database(record):
    try:
        self.db.save(record)
    except DatabaseConnectionError as e:
        # 连接问题：可以重试
        raise ServiceUnavailableError("数据库暂时不可用") from e
    except ConstraintViolationError as e:
        # 数据约束问题：业务逻辑错误
        raise InvalidDataError("数据不满足约束条件") from e
    except PermissionDeniedError as e:
        # 权限问题：需要提升权限或提示用户
        raise AuthorizationError("没有操作权限") from e

合理的异常捕获金字塔

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
def application_entry_point():
    """应用入口点 - 可以捕获所有异常"""
    try:
        main()
    except KeyboardInterrupt:
        logger.info("用户中断程序")
        graceful_shutdown()
    except SystemExit:
        # 允许正常退出
        pass
    except Exception as e:
        logger.exception("未处理的异常导致程序退出")
        show_friendly_error_to_user(e)
        sys.exit(1)

def api_endpoint_handler(request):
    """API端点 - 捕获并转换异常"""
    try:
        return process_request(request)
    except AuthenticationError as e:
        return json_response({"error": "认证失败"}, status=401)
    except PermissionError as e:
        return json_response({"error": "权限不足"}, status=403)
    except ValidationError as e:
        return json_response({"error": "请求数据无效"}, status=400)
    except BusinessLogicError as e:
        return json_response({"error": "业务逻辑错误"}, status=422)
    except Exception as e:
        logger.exception("API内部错误")
        return json_response({"error": "服务器内部错误"}, status=500)

def business_method():
    """业务方法 - 选择性捕获异常"""
    try:
        return perform_business_operation()
    except ExternalServiceError as e:
        # 外部服务错误：可以重试或降级
        return fallback_operation()
    # 其他异常继续传播

3. 提供有意义的错误信息

核心思想

异常信息应该帮助开发者快速定位问题，帮助用户理解发生了什么。

错误示范

1
2
3
4
5
6
7
8
9
10
11
12
# ❌ 无意义的错误信息
def divide(a, b):
    if b == 0:
        raise ValueError("错误发生")  # 什么错误？
    return a / b

# ❌ 泄露技术细节给用户
def create_user(username):
    if len(username) < 3:
        raise Exception(
            "字符串长度必须大于等于3，当前长度：2，内容：'ab'"
        )  # 用户不需要知道这些细节

正确做法

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
# ✅ 提供有意义的错误信息
class DivisionByZeroError(ValueError):
    """除零错误"""
    pass

def divide(a, b):
    if b == 0:
        raise DivisionByZeroError(
            f"除数不能为零: {a} / {b}"
        )
    return a / b

# ✅ 区分技术信息和用户信息
class UsernameValidationError(ValueError):
    def __init__(self, username, reason):
        self.username = username
        self.reason = reason
        # 技术信息，用于日志
        technical_message = f"用户名验证失败: {username}, 原因: {reason}"
        # 用户信息，友好提示
        user_message = self._get_user_friendly_message(reason)
        super().__init__(technical_message)
    
    def _get_user_friendly_message(self, reason):
        messages = {
            "too_short": "用户名至少需要3个字符",
            "too_long": "用户名不能超过20个字符", 
            "invalid_chars": "用户名只能包含字母、数字和下划线"
        }
        return messages.get(reason, "用户名无效")

def validate_username(username):
    if len(username) < 3:
        raise UsernameValidationError(username, "too_short")
    if len(username) > 20:
        raise UsernameValidationError(username, "too_long")
    if not re.match(r'^\w+$', username):
        raise UsernameValidationError(username, "invalid_chars")

上下文丰富的异常

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
class PaymentProcessingError(Exception):
    """支付处理错误"""
    
    def __init__(self, payment_id, amount, currency, reason, user_message=None):
        self.payment_id = payment_id
        self.amount = amount
        self.currency = currency
        self.reason = reason
        self.user_message = user_message or self._default_user_message()
        
        technical_message = (
            f"支付处理失败 [ID: {payment_id}, "
            f"金额: {amount} {currency}, 原因: {reason}]"
        )
        super().__init__(technical_message)
    
    def _default_user_message(self):
        return "支付处理失败，请稍后重试或联系客服"

def process_payment(payment_data):
    try:
        # 支付处理逻辑
        if payment_data['amount'] > payment_data['balance']:
            raise PaymentProcessingError(
                payment_id=payment_data['id'],
                amount=payment_data['amount'],
                currency=payment_data['currency'],
                reason="余额不足",
                user_message="您的账户余额不足"
            )
    except ExternalPaymentGatewayError as e:
        raise PaymentProcessingError(
            payment_id=payment_data['id'],
            amount=payment_data['amount'],
            currency=payment_data['currency'],
            reason=f"支付网关错误: {e}",
            user_message="支付服务暂时不可用，请稍后重试"
        ) from e

4. 记录异常用于调试

核心思想

异常处理应该包含适当的日志记录，便于问题追踪和调试。

错误示范

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
# ❌ 没有记录异常
def process_request(request):
    try:
        return handle_request(request)
    except Exception as e:
        # 只是打印，没有记录到日志系统
        print(f"错误: {e}")
        return error_response("操作失败")

# ❌ 记录了但没有足够信息
def save_data(data):
    try:
        database.save(data)
    except Exception as e:
        logger.error("保存数据失败")  # 没有异常详情和上下文

正确做法

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
# ✅ 完整的异常日志记录
import logging
import traceback

logger = logging.getLogger(__name__)

def process_request(request):
    request_id = generate_request_id()
    logger.info(f"开始处理请求 {request_id}", extra={"request": request})
    
    try:
        result = handle_request(request)
        logger.info(f"请求处理成功 {request_id}")
        return result
        
    except ValidationError as e:
        # 预期内的业务异常，记录警告级别
        logger.warning(
            f"请求数据验证失败 {request_id}",
            extra={"request": request, "error": str(e)},
            exc_info=True  # 包含堆栈跟踪
        )
        return validation_error_response(e)
        
    except ExternalServiceError as e:
        # 外部服务错误，记录错误级别
        logger.error(
            f"外部服务调用失败 {request_id}",
            extra={
                "request": request, 
                "service": e.service_name,
                "operation": e.operation
            },
            exc_info=True
        )
        return service_unavailable_response()
        
    except Exception as e:
        # 未预期的异常，记录严重错误
        logger.critical(
            f"未预期的系统错误 {request_id}",
            extra={
                "request": request,
                "exception_type": type(e).__name__
            },
            exc_info=True  # 这会记录完整的堆栈跟踪
        )
        return internal_error_response()

结构化日志记录

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
class RequestContext:
    """请求上下文，用于关联日志"""
    def __init__(self, request_id, user_id, session_id):
        self.request_id = request_id
        self.user_id = user_id
        self.session_id = session_id

def with_logging_context(context: RequestContext):
    """装饰器：为函数添加日志上下文"""
    def decorator(func):
        def wrapper(*args, **kwargs):
            old_context = getattr(logging, 'context', None)
            logging.context = context
            
            try:
                return func(*args, **kwargs)
            finally:
                logging.context = old_context
        return wrapper
    return decorator

@with_logging_context(context)
def process_order(order_data):
    try:
        validate_order(order_data)
        process_payment(order_data)
        update_inventory(order_data)
        send_confirmation(order_data)
        
    except InsufficientStockError as e:
        logger.warning(
            "库存不足",
            extra={
                "product_id": e.product_id,
                "requested": e.requested,
                "available": e.available
            }
        )
        raise
        
    except PaymentDeclinedError as e:
        logger.error(
            "支付被拒绝",
            extra={
                "order_id": order_data['id'],
                "payment_method": order_data['payment_method'],
                "reason": e.reason
            },
            exc_info=True
        )
        raise
        
    except Exception as e:
        logger.critical(
            "订单处理未知错误",
            extra={"order_data": order_data},
            exc_info=True
        )
        raise

5. 保持异常的纯净性（不要用异常做流程控制）

核心思想

异常应该用于处理真正的异常情况，而不是作为正常的程序流程控制机制。

错误示范

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
# ❌ 使用异常控制流程
def find_user(users, user_id):
    for user in users:
        if user.id == user_id:
            return user
    raise UserNotFoundError()  # 这不是异常情况，是正常逻辑

def process_users():
    try:
        user = find_user(users, 123)
        # 处理用户
    except UserNotFoundError:
        # 用户不存在时的处理
        create_default_user(123)

# ❌ 使用异常进行业务逻辑分支
def calculate_discount(order):
    try:
        if order.amount < 0:
            raise NegativeAmountError()
        # 计算逻辑...
    except NegativeAmountError:
        return 0  # 使用异常决定返回值

正确做法

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
# ✅ 使用返回值控制正常流程
def find_user(users, user_id):
    """查找用户，返回Optional类型明确表示可能找不到"""
    for user in users:
        if user.id == user_id:
            return user
    return None  # 正常情况：没找到

def get_user_or_default(users, user_id):
    """获取用户或返回默认用户"""
    user = find_user(users, user_id)
    if user is None:
        logger.info(f"用户 {user_id} 不存在，创建默认用户")
        return create_default_user(user_id)
    return user

# ✅ 使用特定值或状态对象
class FindResult:
    """查找结果封装"""
    def __init__(self, found, value=None, reason=None):
        self.found = found
        self.value = value
        self.reason = reason
    
    @classmethod
    def success(cls, value):
        return cls(found=True, value=value)
    
    @classmethod
    def not_found(cls, reason):
        return cls(found=False, reason=reason)

def find_user_advanced(users, user_id):
    for user in users:
        if user.id == user_id:
            return FindResult.success(user)
    return FindResult.not_found("用户ID不存在")

# 使用方式
result = find_user_advanced(users, 123)
if result.found:
    process_user(result.value)
else:
    logger.info(f"用户查找失败: {result.reason}")
    handle_user_not_found()

异常的正确使用场景

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
# ✅ 真正的异常情况：数据损坏
def load_configuration():
    config = read_config_file()
    if not validate_config_schema(config):
        # 配置文件损坏是异常情况
        raise CorruptedConfigError("配置文件模式验证失败")
    return config

# ✅ 真正的异常情况：系统资源不可用
def connect_to_database():
    try:
        return create_connection()
    except ConnectionRefusedError as e:
        # 数据库连接失败是异常情况
        raise DatabaseUnavailableError("数据库无法连接") from e

# ✅ 真正的异常情况：违反业务规则
def withdraw_money(account, amount):
    if amount <= 0:
        # 业务规则违反是异常情况
        raise InvalidAmountError("取款金额必须大于零")
    
    if amount > account.balance:
        # 业务规则违反是异常情况
        raise InsufficientFundsError("余额不足")
    
    account.balance -= amount
    return account.balance

流程控制 vs 异常情况总结

场景 正确做法 错误做法
用户不存在 返回None或特定结果对象 抛出UserNotFoundError
数据验证失败 返回验证结果对象 抛出验证异常作为流程控制
可选功能未启用 返回False或特定状态 抛出功能未启用异常
搜索无结果 返回空集合或None 抛出无结果异常
文件不存在（业务预期内） 返回None或创建文件 抛出文件不存在异常
文件损坏（意外情况） 抛出异常 静默返回默认值
数据库连接失败 抛出异常 返回None
权限检查失败 返回False或特定状态 抛出权限异常作为流程控制

通过遵循这些原则，你可以创建出更加健壮、可维护和易于调试的异常处理机制。

宇宙山河浪漫，赞赏动力无限

Welcome to my other publishing channels