import asyncio
from contextlib import AsyncExitStack
#定义一个异步上下文管理函数
class AsyncContextExample:
def __init__(self, name):
self.name = name
async def __aenter__(self):
print(f"Entering {self.name}")
return self
async def __aexit__(self, exc_type, exc_val, exc_tb):
print(f"Exiting {self.name}")
async def main():
stack = AsyncExitStack() # s声明一个上下文管理器
try:
ctx1 = await stack.enter_async_context(AsyncContextExample("Context 1"))
ctx2 = await stack.enter_async_context(AsyncContextExample("Context 2"))
await asyncio.sleep(1)
finally:
await stack.aclose() # 释放全部异步上下文。
if __name__ == "__main__":
asyncio.run(main())
import asyncio
from functools import wraps
from contextlib import AsyncExitStack
# 定义一个异步上下文管理类(保持不变)
class AsyncContextExample:
def __init__(self, name):
self.name = name
async def __aenter__(self):
print(f"Entering {self.name}")
return self
async def __aexit__(self, exc_type, exc_val, exc_tb):
print(f"Exiting {self.name}")
# 定义异步上下文管理装饰器
def async_context_manager(*contexts):
def decorator(func):
@wraps(func)
async def wrapper(*args, **kwargs):
async with AsyncExitStack() as stack:
# 将所有上下文加入stack并获取实例
context_instances = [
await stack.enter_async_context(context)
for context in contexts
]
# 执行被装饰的函数,传入上下文实例
result = await func(*context_instances, *args, **kwargs)
return result
return wrapper
return decorator
# 使用装饰器的新main函数
@async_context_manager(
AsyncContextExample("Context 1"),
AsyncContextExample("Context 2")
)
async def main(ctx1, ctx2):
await asyncio.sleep(1)
if __name__ == "__main__":
asyncio.run(main())