juju/loop.py

   1 import asyncio
   2 import signal
   3
   4
   5 def run(*steps):
   6     """
   7     Helper to run one or more async functions synchronously, with graceful
   8     handling of SIGINT / Ctrl-C.
   9
  10     Returns the return value of the last function.
  11     """
  12     if not steps:
  13         return
  14
  15     task = None
  16     run._sigint = False  # function attr to allow setting from closure
  17     loop = asyncio.get_event_loop()
  18
  19     def abort():
  20         task.cancel()
  21         run._sigint = True
  22
  23     loop.add_signal_handler(signal.SIGINT, abort)
  24     try:
  25         for step in steps:
  26             task = loop.create_task(step)
  27             loop.run_until_complete(asyncio.wait([task], loop=loop))
  28             if run._sigint:
  29                 raise KeyboardInterrupt()
  30             if task.exception():
  31                 raise task.exception()
  32         return task.result()
  33     finally:
  34         loop.remove_signal_handler(signal.SIGINT)