Pythonの `futures`のための` DummyExecutor`

Question

Pythonのfuturesパッケージでは、並行してタスクを実行するためにThreadPoolExecutorとProcessPoolExecutorを楽しむことができます。

しかし、デバッグでは、スレッドやプロセスを生成することなく、真の並列性を一時的にメインのスレッドで逐次実行するダミーのものに一時的に置き換えると便利なことがあります。

DummyExecutorの実装はどこにありますか？このような

Answer 1

何かがそれを行う必要があります。

from concurrent.futures import Future, Executor 
from threading import Lock 


class DummyExecutor(Executor): 

    def __init__(self): 
     self._shutdown = False 
     self._shutdownLock = Lock() 

    def submit(self, fn, *args, **kwargs): 
     with self._shutdownLock: 
      if self._shutdown: 
       raise RuntimeError('cannot schedule new futures after shutdown') 

      f = Future() 
      try: 
       result = fn(*args, **kwargs) 
      except BaseException as e: 
       f.set_exception(e) 
      else: 
       f.set_result(result) 

      return f 

    def shutdown(self, wait=True): 
     with self._shutdownLock: 
      self._shutdown = True 


if __name__ == '__main__': 

    def fnc(err): 
     if err: 
      raise Exception("test") 
     else: 
      return "ok" 

    ex = DummyExecutor() 
    print(ex.submit(fnc, True)) 
    print(ex.submit(fnc, False)) 
    ex.shutdown() 
    ex.submit(fnc, True) # raises exception 

ロックは、おそらくこのような場合には必要ありませんが、それを持って傷つけることはできません。