我要投稿

Python上下文管理器深度解析：从基础到高级应用实战

文章小助手 python

2025-07-11 0 716

Python上下文管理器深度解析：从基础到高级应用实战

一、上下文管理器核心概念

上下文管理器是Python中管理资源的优雅方式，通过__enter__和__exit__方法实现：

class DatabaseConnection:
    def __enter__(self):
        print("建立数据库连接")
        self.conn = "模拟的连接对象"
        return self.conn
    
    def __exit__(self, exc_type, exc_val, exc_tb):
        print("关闭数据库连接")
        self.conn = None

# 使用示例
with DatabaseConnection() as conn:
    print(f"使用连接执行操作: {conn}")

二、contextlib模块的妙用

使用contextlib.contextmanager装饰器快速创建上下文管理器：

from contextlib import contextmanager
import time

@contextmanager
def timer(name):
    start = time.perf_counter()
    try:
        yield
    finally:
        duration = time.perf_counter() - start
        print(f"{name}耗时: {duration:.2f}秒")

# 使用示例
with timer("数据计算"):
    sum(range(1000000))

三、多上下文管理器嵌套

Python支持同时管理多个资源：

class FileHandler:
    def __init__(self, filename, mode):
        self.filename = filename
        self.mode = mode
    
    def __enter__(self):
        self.file = open(self.filename, self.mode)
        print(f"打开文件: {self.filename}")
        return self.file
    
    def __exit__(self, exc_type, exc_val, exc_tb):
        self.file.close()
        print(f"关闭文件: {self.filename}")

# 同时处理两个文件
with FileHandler("source.txt", "r") as src, 
     FileHandler("target.txt", "w") as dst:
    dst.write(src.read())

四、异常处理的正确姿势

上下文管理器能优雅处理异常：

class Transaction:
    def __enter__(self):
        print("开始事务")
        return self
    
    def __exit__(self, exc_type, exc_val, exc_tb):
        if exc_type is None:
            print("提交事务")
        else:
            print(f"回滚事务，异常: {exc_val}")
        return False  # 不抑制异常

# 测试用例
with Transaction():
    print("执行数据库操作")
    # raise ValueError("模拟错误")  # 取消注释测试异常情况

五、异步上下文管理器

Python 3.7+支持异步上下文管理：

import aiohttp
import asyncio

class AsyncWebFetcher:
    async def __aenter__(self):
        self.session = aiohttp.ClientSession()
        return self
    
    async def __aexit__(self, exc_type, exc_val, exc_tb):
        await self.session.close()
    
    async def fetch(self, url):
        async with self.session.get(url) as response:
            return await response.text()

async def main():
    async with AsyncWebFetcher() as fetcher:
        content = await fetcher.fetch("https://example.com")
        print(f"获取内容长度: {len(content)}")

asyncio.run(main())