Python中如何优雅避免异常处理代码重复？

嘿，针对你遇到的「不想重复写一堆异常处理代码」的问题，我给你几个Pythonic的解决方案，结合你提到的异步Socket读写场景（要处理ConnectionResetError、BrokenPipeError、TimeoutError这些），咱们一步步来优化～

先说说你当前的解法：你把异常处理逻辑抽成了单独的函数，这已经是个很好的开端了！不过还有更灵活、更贴合Python风格的方式，让代码更清爽：

方案一：用装饰器统一处理异常（同步/异步都支持） #

装饰器是Python里复用代码的神器之一，不管是同步函数还是异步函数，都可以用装饰器把异常处理逻辑“贴”到需要的函数上，不用每个函数都写一遍try-except。

同步场景示例：

def handle_socket_exceptions(handler_func=None):
    def decorator(func):
        def wrapper(*args, **kwargs):
            try:
                return func(*args, **kwargs)
            except (ConnectionResetError, BrokenPipeError, TimeoutError, ValueError) as e:
                # 调用通用异常处理逻辑
                if handler_func:
                    handler_func(e)
                else:
                    print(f"捕获到异常: {type(e).__name__} - {e}")
                    # 针对特定异常的自定义处理
                    if isinstance(e, ValueError):
                        print("执行ValueError的专属处理逻辑")
                    # 若需重新抛出未匹配的异常，可取消下面注释
                    # else:
                    #     raise e
        return wrapper
    return decorator

# 给需要处理异常的函数加装饰器
@handle_socket_exceptions()
def sync_socket_operation():
    # 模拟同步Socket操作，比如读取数据
    2 / 0  # 触发异常测试

sync_socket_operation()

异步场景示例（适配你的异步Socket场景）：

考虑到你提到了异步Socket，专门写个异步版本的装饰器：

import asyncio

def async_handle_socket_exceptions(handler_func=None):
    def decorator(func):
        async def wrapper(*args, **kwargs):
            try:
                return await func(*args, **kwargs)
            except (ConnectionResetError, BrokenPipeError, TimeoutError, ValueError) as e:
                if handler_func:
                    # 兼容同步/异步处理函数
                    if asyncio.iscoroutinefunction(handler_func):
                        await handler_func(e)
                    else:
                        handler_func(e)
                else:
                    print(f"异步操作捕获异常: {type(e).__name__} - {e}")
                    if isinstance(e, TimeoutError):
                        print("触发超时，执行重试逻辑...")
                    # 无需重新抛出则注释raise
                    # raise e
        return wrapper
    return decorator

# 异步函数使用装饰器
@async_handle_socket_exceptions()
async def async_socket_read():
    # 模拟异步Socket读取操作
    await asyncio.sleep(0.1)
    raise ConnectionResetError("连接被远端重置")

asyncio.run(async_socket_read())

方案二：用上下文管理器封装异常处理（代码块级复用） #

如果是某一段代码块需要统一处理异常，用上下文管理器会更灵活，通过with语句就能把代码块包裹进去，自动应用异常处理逻辑。

同步上下文管理器示例：

from contextlib import contextmanager

@contextmanager
def socket_exception_handler():
    try:
        yield
    except (ConnectionResetError, BrokenPipeError, TimeoutError) as e:
        print(f"捕获到Socket异常: {type(e).__name__} - {e}")
        # 针对不同异常的自定义处理
        if isinstance(e, TimeoutError):
            print("执行超时重试逻辑...")
    except ValueError as e:
        print(f"捕获到值异常: {e}")
    # 无需重新抛出则省略raise

# 使用方式
with socket_exception_handler():
    # 这里写你的Socket操作代码
    2 / 0  # 模拟异常触发

异步上下文管理器示例：

异步场景下用asynccontextmanager实现：

from contextlib import asynccontextmanager
import asyncio

@asynccontextmanager
async def async_socket_exception_handler():
    try:
        yield
    except (ConnectionResetError, BrokenPipeError, TimeoutError) as e:
        print(f"异步Socket异常: {type(e).__name__} - {e}")
        # 模拟异步处理操作，比如异步日志记录
        await asyncio.sleep(0.1)

# 异步使用方式
async def main():
    async with async_socket_exception_handler():
        raise BrokenPipeError("通信管道已断开")

asyncio.run(main())

方案三：优化现有异常处理函数，提升通用性 #

如果你觉得装饰器和上下文管理器有点“重”，也可以优化你当前的函数，让它支持自定义异常类型和处理逻辑，更灵活适配不同场景：

def handle_common_exceptions(e, handled_exceptions=None, exception_handlers=None):
    # 默认要处理的异常类型
    default_handled = (ConnectionResetError, BrokenPipeError, TimeoutError, ValueError)
    handled = handled_exceptions or default_handled
    # 默认异常处理逻辑映射
    default_handlers = {
        ValueError: lambda e: print(f"处理ValueError: {e}"),
        TimeoutError: lambda e: print(f"处理超时异常: {e}")
    }
    handlers = exception_handlers or default_handlers

    print(f'{type(e)=} {e=}')
    if isinstance(e, handled):
        # 执行对应异常的处理逻辑
        handler = handlers.get(type(e), lambda e: print(f"处理未知异常类型: {type(e).__name__}"))
        handler(e)
    else:
        # 非目标异常重新抛出
        raise e

# 使用方式
try:
    # 你的Socket操作代码
    2 / 0
except Exception as e:
    handle_common_exceptions(e)

总结一下 #

• 如果是整个函数都需要统一异常处理，优先用装饰器，代码最简洁，复用性最高；
• 如果是某段代码块需要处理，用上下文管理器，灵活性更强；
• 如果只想小幅改动现有代码，就用通用异常处理函数，改动成本最低。

这些方式都能帮你彻底干掉重复的异常处理代码，让你的代码更干净、更符合Python风格😉

备注：内容来源于stack exchange，提问作者WebOrCode