当前位置：首页 > news >正文

上下文管理器和 contextlib 模块

news 2025/7/15 8:57:09

在 Python 中，上下文管理器（Context Manager）是一种用于管理资源获取与释放的结构，核心作用是自动化资源的进入和退出（常用于文件操作、数据库连接、线程锁、事务处理等）。它的本质是定义好 enter / exit（同步）或 aenter / aexit（异步）方法。

1. 基本概念

同步上下文管理器（with）

class MyResource:def __enter__(self):print("获取资源")return selfdef __exit__(self, exc_type, exc_val, exc_tb):print("释放资源")with MyResource() as r:print("处理中")

输出：

获取资源
处理中
释放资源

异步上下文管理器（async with）

适用于异步资源（如：异步数据库、aiofiles、aiohttp 等）：

class MyAsyncResource:async def __aenter__(self):print("异步获取资源")return selfasync def __aexit__(self, exc_type, exc_val, exc_tb):print("异步释放资源")async with MyAsyncResource() as r:print("异步处理中")

2. contextlib 模块

contextlib 是 Python 标准库中的一个实用工具模块，专门用于简化上下文管理器(context manager)的创建和使用。

主要功能

简化上下文管理器的创建 - 无需编写完整的类实现 enter 和 exit 方法
提供装饰器和工具函数 - 用于常见上下文管理场景
支持可重用的上下文管理器

核心组件

1. @contextmanager 装饰器

基本用法:

from contextlib import contextmanager@contextmanager
def 管理资源(*args, **kwargs):# 这部分相当于 __enter__ 方法资源 = 获取资源(*args, **kwargs)try:yield 资源  # 在这里返回资源给as变量finally:# 这部分相当于 __exit__ 方法释放资源(资源)# 使用方式
with 管理资源() as 资源:print(资源)

示例：

from contextlib import contextmanager@contextmanager
def open_file(path):f = open(path, 'r')try:yield ffinally:f.close()with open_file("test.txt") as f:print(f.read())

2. @asynccontextmanager（异步）

基本用法：

from contextlib import asynccontextmanager@asynccontextmanager
async def async_resource_manager():# 初始化部分 (相当于 __aenter__)resource = await acquire_resource_async()try:yield resource  # 在这里暂停，将资源提供给 with 块finally:# 清理部分 (相当于 __aexit__)await release_resource_async(resource)# 使用方式
async def main():async with async_resource_manager() as res:await do_something_with(res)

特点：

专为异步代码设计：与普通 @contextmanager 不同，它处理的是异步上下文协议 (aenter/aexit)
必须用 async with 调用：不能与普通 with 语句混用
支持异步清理：finally 块中可以执行 await 操作

示例：

from contextlib import asynccontextmanager
import aiofiles@asynccontextmanager
async def open_file_async(path):f = await aiofiles.open(path, 'r')try:yield ffinally:await f.close()async def run():async with open_file_async("test.txt") as f:content = await f.read()

3. 高级实践：FastAPI 启动生命周期封装

from fastapi import FastAPI
from contextlib import asynccontextmanager
from motor.motor_asyncio import AsyncIOMotorClient@asynccontextmanager
async def lifespan(app: FastAPI):app.state.mongo = AsyncIOMotorClient("mongodb://localhost:27017")["mydb"]yieldapp.state.mongo.client.close()app = FastAPI(lifespan=lifespan)