Python学习--上下文管理详解

with EXPR as VAR:    BLOCK# 等价于：manager = EXPRVAR = manager.__enter__()try:    BLOCKfinally:    manager.__exit__(exc_type, exc_val, exc_tb)

class SimpleContext:    """最简单的上下文管理器"""    def __enter__(self):        print("进入上下文")        return self  # 返回的对象会绑定到 as 变量    def __exit__(self, exc_type, exc_val, exc_tb):        print("退出上下文")        # 返回 False 表示异常需要继续传播        # 返回 True 表示异常已被处理        return False    def do_something(self):        print("执行操作")# 使用with SimpleContext() as ctx:    ctx.do_something()print("上下文已退出")

class ManagedFile:    """管理文件操作"""    def __init__(self, filename, mode='r'):        self.filename = filename        self.mode = mode        self.file = None    def __enter__(self):        print(f"打开文件: {self.filename}")        self.file = open(self.filename, self.mode)        return self.file    def __exit__(self, exc_type, exc_val, exc_tb):        if self.file:            self.file.close()            print(f"关闭文件: {self.filename}")        if exc_type is not None:            print(f"发生异常: {exc_type.__name__}: {exc_val}")        # 返回 False 让异常继续传播        return False# 使用print("=== 正常情况 ===")with ManagedFile('test.txt', 'w') as f:    f.write('Hello, World!')    print("文件写入完成")print("\n=== 异常情况 ===")try:    with ManagedFile('test.txt', 'r') as f:        content = f.read()        x = 1 / 0  # 引发异常        print(content)except ZeroDivisionError:    print("捕获到除零异常")

class Connection:    """数据库连接"""    def __init__(self, host):        self.host = host        self.connected = False    def connect(self):        print(f"连接到 {self.host}")        self.connected = True        return self    def close(self):        if self.connected:            print(f"断开连接 {self.host}")            self.connected = False    def query(self, sql):        if not self.connected:            raise RuntimeError("未连接到数据库")        print(f"执行查询: {sql}")        return f"结果: {sql}"class ConnectionContext:    """连接上下文管理器"""    def __init__(self, host):        self.host = host        self.connection = None    def __enter__(self):        # 创建并返回连接对象        self.connection = Connection(self.host)        self.connection.connect()        return self.connection    def __exit__(self, exc_type, exc_val, exc_tb):        if self.connection:            self.connection.close()# 使用with ConnectionContext('localhost') as conn:    result = conn.query('SELECT * FROM users')    print(result)

class NestedContext:    """嵌套上下文演示"""    def __init__(self, name):        self.name = name    def __enter__(self):        print(f"进入 {self.name}")        return self    def __exit__(self, exc_type, exc_val, exc_tb):        print(f"退出 {self.name}")        return False    def do_work(self):        print(f"在 {self.name} 中工作")# 嵌套使用with NestedContext('外层') as outer:    outer.do_work()    with NestedContext('内层') as inner:        inner.do_work()        with NestedContext('最内层') as innermost:            innermost.do_work()    outer.do_work()

class ExceptionHandler:    """异常处理器"""    def __init__(self, handle_exceptions=True):        self.handle_exceptions = handle_exceptions    def __enter__(self):        print("进入异常处理器")        return self    def __exit__(self, exc_type, exc_val, exc_tb):        """        exc_type: 异常类型        exc_val: 异常实例        exc_tb: 回溯信息        """        if exc_type is not None:            print(f"捕获到异常: {exc_type.__name__}: {exc_val}")            if self.handle_exceptions:                print("异常已被处理")                return True  # 抑制异常        return False  # 不抑制异常# 使用print("=== 抑制异常 ===")with ExceptionHandler(True):    print("执行可能出错的代码")    x = 1 / 0    print("这行不会执行")print("\n=== 不抑制异常 ===")try:    with ExceptionHandler(False):        print("执行可能出错的代码")        x = 1 / 0except ZeroDivisionError:    print("外部捕获到异常")

import timeimport randomclass RetryContext:    """重试上下文管理器"""    def __init__(self, max_retries=3, delay=1):        self.max_retries = max_retries        self.delay = delay        self.retry_count = 0    def __enter__(self):        print(f"开始操作（最多重试 {self.max_retries} 次）")        return self    def __exit__(self, exc_type, exc_val, exc_tb):        if exc_type is None:            return False  # 没有异常，正常退出        self.retry_count += 1        print(f"操作失败 ({exc_type.__name__}), 重试 {self.retry_count}/{self.max_retries}")        if self.retry_count < self.max_retries:            time.sleep(self.delay)            return True  # 抑制异常，触发重试        print("达到最大重试次数")        return False  # 不抑制异常，让异常继续传播def unstable_operation():    """不稳定操作"""    if random.random() < 0.7:  # 70% 概率失败        raise ConnectionError("网络连接失败")    return "操作成功"# 使用重试retry = RetryContext(max_retries=3, delay=0.5)for attempt in range(5):    print(f"\n尝试 {attempt + 1}:")    with retry as r:        result = unstable_operation()        print(f"成功: {result}")

import timeclass Timer:    """计时器上下文管理器"""    def __enter__(self):        self.start = time.perf_counter()        return self    def __exit__(self, exc_type, exc_val, exc_tb):        self.end = time.perf_counter()        self.elapsed = self.end - self.start        print(f"耗时: {self.elapsed:.6f} 秒")        if exc_type is not None:            print(f"发生异常: {exc_type.__name__}")        return False    def __str__(self):        return f"Timer({self.elapsed:.6f}s)" if hasattr(self, 'elapsed') else "Timer(未开始)"# 使用with Timer() as timer:    time.sleep(0.5)    print("执行耗时操作")with Timer():    total = 0    for i in range(1000000):        total += i    print(f"计算结果: {total}")

class MultiContext:    """同时管理多个资源的上下文"""    def __init__(self):        self.resources = []    def add_resource(self, name):        self.resources.append(name)        return self    def __enter__(self):        print("打开所有资源:")        for resource in self.resources:            print(f"  - 打开 {resource}")        return self.resources    def __exit__(self, exc_type, exc_val, exc_tb):        print("关闭所有资源:")        for resource in reversed(self.resources):            print(f"  - 关闭 {resource}")        if exc_type:            print(f"发生异常: {exc_type.__name__}")        return False# 使用with MultiContext().add_resource('数据库').add_resource('文件').add_resource('网络连接') as resources:    print(f"使用资源: {resources}")    # 执行操作print("\n=== 同时打开多个文件 ===")with open('file1.txt', 'w') as f1, open('file2.txt', 'w') as f2:    f1.write('Hello')    f2.write('World')

class ContextStack:    """上下文堆栈"""    def __init__(self):        self.stack = []        self.depth = 0    def __enter__(self):        self.depth += 1        self.stack.append(f"级别{self.depth}")        print(f"进入上下文级别 {self.depth}")        return self    def __exit__(self, exc_type, exc_val, exc_tb):        print(f"退出上下文级别 {self.depth}")        self.stack.pop()        self.depth -= 1        if exc_type:            print(f"  异常: {exc_type.__name__}")        return False    def get_stack(self):        return self.stack.copy()# 使用堆栈stack = ContextStack()with stack as s1:    print(f"当前堆栈: {stack.get_stack()}")    with stack as s2:        print(f"当前堆栈: {stack.get_stack()}")        with stack as s3:            print(f"当前堆栈: {stack.get_stack()}")        print(f"退出内层后: {stack.get_stack()}")    print(f"退出中层后: {stack.get_stack()}")

import asyncioclass AsyncResource:    """异步资源管理器"""    def __init__(self, name):        self.name = name        self.is_open = False    async def __aenter__(self):        print(f"异步打开 {self.name}")        await asyncio.sleep(0.1)  # 模拟异步操作        self.is_open = True        return self    async def __aexit__(self, exc_type, exc_val, exc_tb):        print(f"异步关闭 {self.name}")        await asyncio.sleep(0.1)  # 模拟异步操作        self.is_open = False        if exc_type:            print(f"发生异常: {exc_type.__name__}")        return False    async def work(self):        if not self.is_open:            raise RuntimeError("资源未打开")        print(f"在 {self.name} 中工作")        await asyncio.sleep(0.2)async def main():    print("=== 异步上下文管理 ===")    async with AsyncResource("数据库连接") as res:        await res.work()        await res.work()    print("资源已关闭")# 运行asyncio.run(main())

import asynciofrom typing import List, Optionalclass AsyncConnection:    """异步数据库连接"""    def __init__(self, conn_id):        self.conn_id = conn_id        self.in_use = False    async def connect(self):        print(f"连接 {self.conn_id} 建立中...")        await asyncio.sleep(0.2)        print(f"连接 {self.conn_id} 已建立")        return self    async def close(self):        print(f"连接 {self.conn_id} 关闭中...")        await asyncio.sleep(0.1)        print(f"连接 {self.conn_id} 已关闭")    async def query(self, sql):        await asyncio.sleep(0.1)        return f"连接 {self.conn_id} 查询结果: {sql}"class AsyncConnectionPool:    """异步连接池"""    def __init__(self, pool_size=3):        self.pool_size = pool_size        self.connections: List[AsyncConnection] = []        self.available: List[AsyncConnection] = []        self.lock = asyncio.Lock()    async def __aenter__(self):        """初始化连接池"""        print(f"初始化连接池 (大小: {self.pool_size})")        for i in range(self.pool_size):            conn = AsyncConnection(f"conn_{i+1}")            await conn.connect()            self.connections.append(conn)            self.available.append(conn)        return self    async def __aexit__(self, exc_type, exc_val, exc_tb):        """关闭所有连接"""        print("关闭连接池...")        for conn in self.connections:            await conn.close()        if exc_type:            print(f"连接池异常: {exc_type.__name__}")        return False    async def get_connection(self) -> AsyncConnection:        """获取可用连接"""        async with self.lock:            while not self.available:                await asyncio.sleep(0.1)            conn = self.available.pop(0)            conn.in_use = True            return conn    async def release_connection(self, conn: AsyncConnection):        """释放连接"""        async with self.lock:            conn.in_use = False            self.available.append(conn)    @asynccontextmanager    async def connection(self):        """获取连接的上下文管理器"""        conn = await self.get_connection()        try:            yield conn        finally:            await self.release_connection(conn)async def worker(pool: AsyncConnectionPool, worker_id: int):    """工作协程"""    async with pool.connection() as conn:        print(f"工作者 {worker_id} 获取到连接 {conn.conn_id}")        result = await conn.query(f"SELECT * FROM table_{worker_id}")        print(f"工作者 {worker_id}: {result}")        await asyncio.sleep(0.3)async def main():    async with AsyncConnectionPool(3) as pool:        # 创建多个工作者        tasks = [worker(pool, i) for i in range(5)]        await asyncio.gather(*tasks)asyncio.run(main())

class CompleteContext:    """完整的上下文管理器示例"""    def __init__(self, resource, auto_cleanup=True):        self.resource = resource        self.auto_cleanup = auto_cleanup        self.acquired = False    def __enter__(self):        """1. 获取资源"""        try:            print(f"获取资源: {self.resource}")            self.resource.acquire()            self.acquired = True            return self.resource        except Exception as e:            print(f"获取资源失败: {e}")            raise    def __exit__(self, exc_type, exc_val, exc_tb):        """2. 释放资源"""        try:            if self.acquired and self.auto_cleanup:                print(f"释放资源: {self.resource}")                self.resource.release()        except Exception as e:            print(f"释放资源失败: {e}")            # 不抛出异常，避免掩盖原始异常        """3. 异常处理"""        if exc_type is not None:            print(f"捕获到异常: {exc_type.__name__}: {exc_val}")            # 根据需要决定是否抑制异常            if self.should_suppress(exc_type):                print("抑制异常")                return True        return False    def should_suppress(self, exc_type):        """决定是否抑制特定异常"""        return False