Python 3에서 비동기 컨텍스트 관리자를 중첩하는 방법

Question

중첩 된 방식으로 함께 일반적으로 사용되는 두 개의 비동기 컨텍스트 관리자가 있지만 일반적으로 두 번째 결과 만 본문에 일반적으로 사용된다고 가정합니다.

async with context_mgr_2() as cm2: 
    ...do something with cm2... 

가 contextlib.nested : 우리가 단지 수 있도록 우리가 이러한 상황에 맞는 관리자 둥지를 단일 문맥 관리자를 만들 수있는 방법

async with context_mgr_1() as cm1: 
    async with cm2.context_mgr_2() as cm2: 
     ...do something with cm2... 

: 예를 들어, 우리는 우리 자신이 많이 입력 발견 비동기 컨텍스트 관리자를 위해이 작업을 수행하는 데 사용되었지만 asyncio에서 이와 같은 도우미를 찾지 못했습니다.

Answer 1

내부적으로이 기능을 사용하여 동기식 및 비동기식 컨텍스트 관리자를 모두 비동기식 관리자로 래핑했습니다. AsyncExitStack 스타일 푸시 의미뿐만 아니라 단순히 여러 관리자를 래핑 할 수 있습니다.

그것은 상당히 잘 테스트,하지만 테스트를 게시하거나이를 지원 계획하고 있지 않다, 그래서 당신의 자신의 위험에 사용 ...

import asyncio 
import logging 
import sys 

from functools import wraps 

class AsyncContextManagerChain(object): 

    def __init__(self, *managers): 
     self.managers = managers 
     self.stack = [] 
     self.values = [] 

    async def push(self, manager): 
     try: 
      if hasattr(manager, '__aenter__'): 
       value = await manager.__aenter__() 
      else: 
       value = manager.__enter__() 

      self.stack.append(manager) 
      self.values.append(value) 
      return value 
     except: 
      # if we encounter an exception somewhere along our enters, 
      # we'll stop adding to the stack, and pop everything we've 
      # added so far, to simulate what would happen when an inner 
      # block raised an exception. 
      swallow = await self.__aexit__(*sys.exc_info()) 
      if not swallow: 
       raise 

    async def __aenter__(self): 
     value = None 

     for manager in self.managers: 
      value = await self.push(manager) 

     return value 

    async def __aexit__(self, exc_type, exc, tb): 
     excChanged = False 
     swallow = False # default value 
     while self.stack: 
      # no matter what the outcome, we want to attempt to call __aexit__ on 
      # all context managers 
      try: 
       swallow = await self._pop(exc_type, exc, tb) 
       if swallow: 
        # if we swallow an exception on an inner cm, outer cms would 
        # not receive it at all... 
        exc_type = None 
        exc = None 
        tb = None 
      except: 
       # if we encounter an exception while exiting, that is the 
       # new execption we send upward 
       excChanged = True 
       (exc_type, exc, tb) = sys.exc_info() 
       swallow = False 

     if exc is None: 
      # when we make it to the end, if exc is None, it was swallowed 
      # somewhere along the line, and we've exited everything successfully, 
      # so tell python to swallow the exception for real 
      return True 
     elif excChanged: 
      # if the exception has been changed, we need to raise it here 
      # because otherwise python will just raise the original exception 
      if not swallow: 
       raise exc 
     else: 
      # we have the original exception still, we just let python handle it... 
      return swallow 

    async def _pop(self, exc_type, exc, tb): 
    manager = self.stack.pop() 
    if hasattr(manager, '__aexit__'): 
     return await manager.__aexit__(exc_type, exc, tb) 
    else: 
     return manager.__exit__(exc_type, exc, tb) 

Answer 2

케빈의 대답은 contextlib.ExitStack 따르지 않는 3.5.2에서는 impl을 사용 했으므로 파이썬 3.5.2의 공식 impl을 기반으로 한 것을 만들었습니다. 나는 문제를 발견하면 impl을 업데이트 할 것이다.

GitHub의 요점 링크 : https://gist.github.com/thehesiod/b8442ed50e27a23524435a22f10c04a0