Python `with`コンテキストとジェネレータ/コルーチン/タスク

Question

python withブロックを使って実験して、そのブロック内のアクションに修飾子を適用したいと思います。しかし、コルーチンの存在下でこれを分かりやすく行うことが可能かどうかはわかりません。例えば

は、私は一時的にこのようなスタックにプッシュしWithContextオブジェクトがあるとします。

class WithContext: 
    stack = [] 
    def __init__(self, val): 
     self.val = val 
    def __enter__(self): 
     WithContext.stack.append(self.val) 
    def __exit__(self, exc_type, exc_val, exc_tb): 
     WithContext.stack.pop() 
def do_scoped_contextual_thing(): 
    print(WithContext.stack[-1]) 

を（明らかにスタックメンバーは、スレッドローカルでなければならないが、今のためにそれを無視します。）

次に、このコードは：

with WithContext("a"): 
    do_scoped_contextual_thing() 
    with WithContext("b"): 
     do_scoped_contextual_thing() 
with WithContext("c"): 
    do_scoped_contextual_thing() 

が印刷されます：

a 
b 
c 

しかし、今、私はコルーチンの状況があるとします。私は出力にこのコードをしたい

def coroutine(): 
    with WithContext("inside"): 
     yield 1 
     do_scoped_contextual_thing() 
     yield 2 

with WithContext("outside"): 
    for e in coroutine(): 
     do_scoped_contextual_thing() 
     print("got " + str(e)) 

を：

outside 
got 1 
inside 
outside 
got 2 

しかし、実際には出力は以下となります。

inside 
got 1 
inside 
inside 
got 2 

外側が__enter__の内側にあるので内側に変わったコルーチンはスタックの上に値を置き、__exit__は、コルーチンが終了するまで呼び出されません（コルーチンの中に出入りする際に常に入り口と出口がありません）。

この問題を回避する方法はありますか？ 「コルーチンローカル」変数はありますか？

Answer 1

半分の壊れた "解決策"は、コンテキストをスタックフレームの位置に関連付け、コンテキストをルックアップするときにその位置を確認することです。

class WithContext: 
    _stacks = defaultdict(list) 

    def __init__(self, val): 
     self.val = val 

    def __enter__(self): 
     _, file, _, method, _, _ = inspect.stack()[1] 
     WithContext._stacks[(file, method)].append(self.val) 

    def __exit__(self, exc_type, exc_val, exc_tb): 
     _, file, _, method, _, _ = inspect.stack()[1] 
     WithContext._stacks[(file, method)].pop() 

    @staticmethod 
    def get_context(): 
     for frame in inspect.stack()[1:]: 
      _, file, _, method, _, _ = frame 
      r = WithContext._stacks[(file, method)] 
      if r: 
       return r[-1] 
     raise ValueError("no context") 

スタックフレームを常に検索するのは、単に値を渡すよりも高価で、これを書いた人には伝えたくないことに注意してください。

これはさらに複雑な状況ではまだ解消されることに注意してください。例えば

：

• 同じ方法で二回スタック上にある場合はどう？
• ジェネレータが1つの場所から少し反復され、次に別の場所からもう少し反復される場合はどうなりますか？
• 再帰的な発電機はどうですか？
• 非同期メソッドはどうですか？

Answer 2

私はこれについては気にしませんが、テストコードを変更して数回コルーチンを再入力しました。 @ CraigGidneyのソリューションと同様に、inspectモジュールを使用して、WithContextオブジェクトが作成されたコールスタック（別名「スコープ」）の情報にアクセスしてキャッシュします。

次に、キャッシュされた値を探してスタックを検索し、id関数を使用して、実際のフレームオブジェクトへの参照を保持しないようにします。

import inspect 

class WithContext: 
    stack = [] 
    frame_to_stack = {} 
    def __init__(self, val): 
     self.val = val 
    def __enter__(self): 
     stk = inspect.stack(context=3) 
     caller_id = id(stk[1].frame) 
     WithContext.frame_to_stack[caller_id] = len(WithContext.stack) 
     WithContext.stack.append((caller_id, self.val)) 

    def __exit__(self, exc_type, exc_val, exc_tb): 
     wc = WithContext.stack.pop() 
     del WithContext.frame_to_stack[wc[0]] 

def do_scoped_contextual_thing(): 
    stack = inspect.stack(context=0) 
    f2s = WithContext.frame_to_stack 

    for f in stack: 
     wcx = f2s.get(id(f.frame)) 

     if wcx is not None: 
      break 
    else: 
     raise ValueError("No context object in scope.") 

    print(WithContext.stack[wcx][1]) 

def coroutine(): 
    with WithContext("inside"): 
     for n in range(3): 
      yield 1 
      do_scoped_contextual_thing() 
      yield 2 

with WithContext("outside"): 
    for e in coroutine(): 
     do_scoped_contextual_thing() 
     print("got " + str(e))