在一系列上下文管理器中可迭代Python的友好用法

Question

我正在寫一個試圖爲調度作業提供持久隊列的小型庫。我的持久性代碼提供了迭代未決作業描述的方法;我還想保證派遣的工作最終會被標記爲已完成或失敗。

要做到這一點，我第一個實現它讓我的用戶可以這樣做：

for c in some_iterator_object: 
    with c as x: 
    ... 

我不喜歡這個解決方案有以下幾個原因。首先，我想從我的隊列中獲取作爲單個操作的作業描述（如果隊列爲空，則失敗），因此採集由迭代器的__next__方法和上下文的__exit__中的發佈完成經理。

爲了確保調用上下文管理器，我的__next__返回一個不能直接替換值的包裝類，所以如果用戶忘記調用上下文管理器，它將拋出一個明確的錯誤。

有沒有辦法將這兩個語句合併爲一個？理想情況下，我想讓用戶做所有，同時能夠攔截for塊的內容引發的異常。

編輯：我發現通過實驗，如果我讓一個未完成的發電機被垃圾收集，產量語句會拋出一個內部異常，所以我可以寫的東西粗像

try: 
    ... 
    success = False 
    yield val 
    success = True 
    ... 
finally: 
    if success: 
    ... 

但是，如果我理解正確，這取決於垃圾收集器的運行情況，而且它似乎是一個我不應該碰到的內部機制。

Answer 1

如果你想自動進入您的上下文的經理，因爲他們正在迭代器返回，您可以編寫自己的迭代器類是這樣的：

class ContextManagersIterator: 

    def __init__(self, it): 
     self._it = iter(it) 
     self._last = None 

    def __iter__(self): 
     return self 

    def __next__(self): 
     self.__exit__(None, None, None) 

     item = next(self._it) 
     item.__enter__() 
     self._last = item 

     return item 

    def __enter__(self): 
     return self 

    def __exit__(self, exc_type, exc_value, exc_traceback): 
     last = self._last 
     if last is not None: 
      self._last = None 
      return last.__exit__(exc_type, exc_value, exc_traceback) 

用法示例：

from contextlib import contextmanager 

@contextmanager 
def my_context_manager(name): 
    print('enter', name) 
    try: 
     yield 
    finally: 
     print('exit', name) 

sequence = [ 
    my_context_manager('x'), 
    my_context_manager('y'), 
    my_context_manager('z'), 
] 

with ContextManagersIterator(sequence) as it: 
    for item in it: 
     print(' work') 

# Output: 
# enter x 
# work 
# exit x 
# enter y 
# work 
# exit y 
# enter z 
# work 
# exit z 

這個ContextManagersIterator類在它們被返回之前就處理其值的調用__enter__。在返回另一個值之前調用__exit__（如果一切順利）或循環中引發異常。