异步上下文示例

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())
使用 Hugo 构建
主题 StackJimmy 设计