Python24_async with 语法

文章目录

• Python24_async with 语法
• • @[toc]
  • 1. 基础概念
  • • 1.1 什么是 `async with`？
    • 1.2 为什么需要 `async with`？
  • 2. 核心原理
  • • 2.1 异步上下文管理器协议
    • 2.2 执行流程
  • 3. 常见使用场景
  • • 3.1 异步文件操作（aiofiles）
    • 3.2 异步数据库连接（aiomysql/aiopg）
    • 3.3 异步 HTTP 请求（aiohttp）
    • 3.4 异步锁（asyncio.Lock）
  • 4. 常见问题与解答（Q&A）
  • • Q1: `async with` 和 `with` 有什么区别？
    • Q2: 如何自定义异步上下文管理器？
    • Q3: `async with` 可以嵌套使用吗？
    • Q4: `__aexit__` 的参数是什么意思？
    • Q5: 可以在 `async with` 中使用 `await` 吗？
    • Q6: 如何正确处理异常？
    • Q7: 异步上下文管理器在 `__aexit__` 中可以再使用 `await` 吗？
  • 5. 最佳实践
  • • 5.1 使用 `@asynccontextmanager` 简化代码
    • 5.2 超时控制
    • 5.3 避免常见错误
  • 6. 速查表
  • 7. 相关知识点延伸

1. 基础概念

1.1 什么是 async with？

async with 是 Python 异步上下文管理器（Asynchronous Context Manager）的语法糖，用于在异步代码中管理资源的获取和释放。

async with 表达式 as 变量:
    # 异步代码块
    pass

1.2 为什么需要 async with？

场景	普通 with	async with
文件操作（同步）	✅ 适用	❌ 不适用
异步数据库连接	❌ 阻塞	✅ 非阻塞
异步网络请求	❌ 阻塞	✅ 非阻塞
异步锁（Lock）	❌ 阻塞	✅ 非阻塞

---

2. 核心原理

2.1 异步上下文管理器协议

一个对象要成为异步上下文管理器，必须实现以下两个异步方法：

class AsyncContextManager:
    async def __aenter__(self):
        """进入上下文时调用"""
        return self
    
    async def __aexit__(self, exc_type, exc_val, exc_tb):
        """退出上下文时调用"""
        # 处理异常或清理资源
        pass

2.2 执行流程

┌─────────────────┐
│   async with    │
│   表达式 as x:   │
└────────┬────────┘
         ▼
┌─────────────────┐
│  await __aenter__() │  ← 异步获取资源
│  返回值赋给 x      │
└────────┬────────┘
         ▼
┌─────────────────┐
│    执行代码块     │
└────────┬────────┘
         ▼
┌─────────────────┐
│ await __aexit__() │  ← 异步释放资源
│ (无论是否异常都执行) │
└─────────────────┘

---

3. 常见使用场景

3.1 异步文件操作（aiofiles）

import aiofiles
import asyncio

async def read_file():
    async with aiofiles.open('data.txt', 'r') as f:
        content = await f.read()
        print(content)

asyncio.run(read_file())

3.2 异步数据库连接（aiomysql/aiopg）

import aiomysql

async def query_db():
    async with aiomysql.create_pool(
        host='localhost', 
        user='root', 
        password='pwd',
        db='test'
    ) as pool:
        async with pool.acquire() as conn:
            async with conn.cursor() as cur:
                await cur.execute("SELECT * FROM users")
                result = await cur.fetchall()
                return result

3.3 异步 HTTP 请求（aiohttp）

import aiohttp
import asyncio

async def fetch_url(url):
    async with aiohttp.ClientSession() as session:
        async with session.get(url) as response:
            return await response.text()

# 使用
html = asyncio.run(fetch_url('https://api.github.com'))

3.4 异步锁（asyncio.Lock）

import asyncio

lock = asyncio.Lock()
counter = 0

async def increment():
    global counter
    async with lock:  # 确保同一时间只有一个协程修改 counter
        current = counter
        await asyncio.sleep(0.1)  # 模拟耗时操作
        counter = current + 1

async def main():
    await asyncio.gather(*[increment() for _ in range(10)])
    print(f"Counter: {counter}")  # 输出: Counter: 10

asyncio.run(main())

---

4. 常见问题与解答（Q&A）

Q1: async with 和 with 有什么区别？

答：

特性	with	async with
方法	__enter__ / __exit__	__aenter__ / __aexit__
调用方式	同步调用	需要 await
使用场景	同步资源管理	异步资源管理
性能	阻塞	非阻塞，可并发

错误示例：

# ❌ 错误：在异步函数中使用同步 with 管理异步资源
async def wrong():
    with aiohttp.ClientSession() as session:  # 错误！
        pass

Q2: 如何自定义异步上下文管理器？

答：

方式一：类实现

class AsyncDatabase:
    async def __aenter__(self):
        self.conn = await create_connection()
        return self.conn
    
    async def __aexit__(self, exc_type, exc, tb):
        await self.conn.close()
        # 返回 True 表示异常已处理，不再传播
        return False

# 使用
async with AsyncDatabase() as conn:
    await conn.query("SELECT 1")

方式二：装饰器（asynccontextmanager）

from contextlib import asynccontextmanager

@asynccontextmanager
async def managed_resource():
    print("获取资源...")
    resource = await create_async_resource()
    try:
        yield resource
    finally:
        print("释放资源...")
        await resource.cleanup()

# 使用
async with managed_resource() as res:
    await res.do_something()

Q3: async with 可以嵌套使用吗？

答： 可以，支持多种写法：

# 方式一：嵌套
async with A() as a:
    async with B() as b:
        pass

# 方式二：单行（Python 3.10+）
async with A() as a, B() as b:
    pass

# 方式三：括号（Python 3.10+，推荐）
async with (
    A() as a,
    B() as b,
    C() as c
):
    pass

Q4: __aexit__ 的参数是什么意思？

答：

async def __aexit__(self, exc_type, exc_val, exc_tb):
    """
    exc_type: 异常类型（如 ValueError）
    exc_val:  异常实例
    exc_tb:   异常追踪信息
    """
    if exc_type is not None:
        print(f"发生异常: {exc_val}")
        # 返回 True 会抑制异常，False 或 None 会传播异常
        return True

示例：

class SuppressError:
    async def __aenter__(self):
        return self
    
    async def __aexit__(self, exc_type, exc_val, exc_tb):
        if exc_type == ValueError:
            print(f"捕获 ValueError: {exc_val}")
            return True  # 抑制异常
        return False  # 其他异常正常抛出

async def test():
    async with SuppressError():
        raise ValueError("测试错误")  # 被捕获，不会抛出
    print("继续执行")

Q5: 可以在 async with 中使用 await 吗？

答： 可以，而且这是常态：

async def process():
    async with get_session() as session:
        result = await session.fetch_data()  # ✅ 正常
        await session.save(result)           # ✅ 正常

Q6: 如何正确处理异常？

答：

class SafeAsyncContext:
    async def __aenter__(self):
        return self
    
    async def __aexit__(self, exc_type, exc_val, exc_tb):
        # 清理资源（无论是否异常都会执行）
        await self.cleanup()
        
        # 选择是否处理异常
        if exc_type is asyncio.CancelledError:
            return True  # 抑制取消异常
        return False  # 其他异常继续传播

# 外部捕获
async def main():
    try:
        async with SafeAsyncContext():
            raise ValueError("测试")
    except ValueError:
        print("捕获到 ValueError")

Q7: 异步上下文管理器在 __aexit__ 中可以再使用 await 吗？

答： 可以，这是设计用途之一：

class AsyncConnection:
    async def __aexit__(self, exc_type, exc_val, exc_tb):
        # 异步关闭连接
        await self.conn.close()
        
        # 异步记录日志
        await log_async("连接已关闭")
        
        # 异步发送指标
        await metrics.send("connection_closed")

---

5. 最佳实践

5.1 使用 @asynccontextmanager 简化代码

from contextlib import asynccontextmanager

@asynccontextmanager
async def transaction(db):
    await db.begin()
    try:
        yield db
        await db.commit()
    except Exception:
        await db.rollback()
        raise

# 使用
async with transaction(database) as db:
    await db.execute("INSERT ...")

5.2 超时控制

import asyncio

async def with_timeout():
    try:
        async with asyncio.timeout(5):  # Python 3.11+
            async with aiohttp.ClientSession() as session:
                async with session.get('https://slow.com') as resp:
                    return await resp.text()
    except asyncio.TimeoutError:
        print("请求超时")

5.3 避免常见错误

# ❌ 错误：忘记 await
async with some_async_context():  # 正确
    pass

with some_async_context():  # 错误！协程没有被 await

# ❌ 错误：在同步函数中使用 async with
def sync_function():
    async with something():  # 错误！同步函数不能有 await
        pass

# ✅ 正确：使用 asyncio.run 或 await
async def async_function():
    async with something():
        pass

---

6. 速查表

需求	代码示例
创建异步上下文管理器	实现 __aenter__ 和 __aexit__
简化创建	使用 @asynccontextmanager
多个资源	async with A() as a, B() as b:
异常处理	在 __aexit__ 中判断 exc_type
抑制异常	__aexit__ 返回 True
超时控制	async with asyncio.timeout(10):

---

7. 相关知识点延伸

• async for - 异步迭代器
• asyncio.gather() - 并发执行多个协程
• asyncio.create_task() - 创建后台任务
• contextlib.AsyncExitStack - 动态管理多个异步上下文

---