a context manager that controls event loop life cycle the context manager always creates a new event loop allows to run async functions inside it and properly finalizes the loop at the context manager exit if debug is true the event loop will be run in debug mode if loopfactory is passed it is used for new event loop creation asyncio runmain debugtrue is a shortcut for with asyncio runnerdebugtrue as runner runner runmain the run method can be called multiple times within the runner s context this can be useful for interactive console e g ipython unittest runners console tools everywhere when async code is called from existing sync framework and where the preferred single asyncio run call doesn t work note the class is final it is not intended for inheritance shutdown and close event loop if self state is not state initialized return try loop self loop cancelalltasksloop loop rununtilcompleteloop shutdownasyncgens loop rununtilcomplete loop shutdowndefaultexecutorconstants threadjointimeout finally if self seteventloop events seteventloopnone loop close self loop none self state state closed def getloopself run a coroutine inside the embedded event loop if not coroutines iscoroutinecoro raise valueerrora coroutine was expected got r formatcoro if events getrunningloop is not none fail fast with short traceback raise runtimeerror runner run cannot be called from a running event loop self lazyinit if context is none context self context task self loop createtaskcoro contextcontext if threading currentthread is threading mainthread and signal getsignalsignal sigint is signal defaultinthandler siginthandler functools partialself onsigint maintasktask try signal signalsignal sigint siginthandler except valueerror signal signal may throw if threading mainthread does not support signals e g embedded interpreter with signals not registered see gh91880 siginthandler none else siginthandler none self interruptcount 0 try return self loop rununtilcompletetask except exceptions cancellederror if self interruptcount 0 uncancel getattrtask uncancel none if uncancel is not none and uncancel 0 raise keyboardinterrupt raise cancellederror finally if siginthandler is not none and signal getsignalsignal sigint is siginthandler signal signalsignal sigint signal defaultinthandler def lazyinitself if self state is state closed raise runtimeerrorrunner is closed if self state is state initialized return if self loopfactory is none self loop events neweventloop if not self seteventloop call seteventloop only once to avoid calling attachloop multiple times on child watchers events seteventloopself loop self seteventloop true else self loop self loopfactory if self debug is not none self loop setdebugself debug self context contextvars copycontext self state state initialized def onsigintself signum frame maintask self interruptcount 1 if self interruptcount 1 and not maintask done maintask cancel wakeup loop if it is blocked by select with long timeout self loop callsoonthreadsafelambda none return raise keyboardinterrupt def runmain debugnone loopfactorynone if events getrunningloop is not none fail fast with short traceback raise runtimeerror asyncio run cannot be called from a running event loop with runnerdebugdebug loopfactoryloopfactory as runner return runner runmain def cancelalltasksloop tocancel tasks alltasksloop if not tocancel return for task in tocancel task cancel loop rununtilcompletetasks gathertocancel returnexceptionstrue for task in tocancel if task cancelled continue if task exception is not none loop callexceptionhandler message unhandled exception during asyncio run shutdown exception task exception task task a context manager that controls event loop life cycle the context manager always creates a new event loop allows to run async functions inside it and properly finalizes the loop at the context manager exit if debug is true the event loop will be run in debug mode if loop_factory is passed it is used for new event loop creation asyncio run main debug true is a shortcut for with asyncio runner debug true as runner runner run main the run method can be called multiple times within the runner s context this can be useful for interactive console e g ipython unittest runners console tools everywhere when async code is called from existing sync framework and where the preferred single asyncio run call doesn t work note the class is final it is not intended for inheritance shutdown and close event loop return embedded event loop run a coroutine inside the embedded event loop fail fast with short traceback signal signal may throw if threading main_thread does not support signals e g embedded interpreter with signals not registered see gh 91880 cancellederror call set_event_loop only once to avoid calling attach_loop multiple times on child watchers wakeup loop if it is blocked by select with long timeout execute the coroutine and return the result this function runs the passed coroutine taking care of managing the asyncio event loop finalizing asynchronous generators and closing the default executor this function cannot be called when another asyncio event loop is running in the same thread if debug is true the event loop will be run in debug mode this function always creates a new event loop and closes it at the end it should be used as a main entry point for asyncio programs and should ideally only be called once the executor is given a timeout duration of 5 minutes to shutdown if the executor hasn t finished within that duration a warning is emitted and the executor is closed example async def main await asyncio sleep 1 print hello asyncio run main fail fast with short traceback
__all__ = ('Runner', 'run')

import contextvars
import enum
import functools
import threading
import signal
from . import coroutines
from . import events
from . import exceptions
from . import tasks
from . import constants

class _State(enum.Enum):
    CREATED = "created"
    INITIALIZED = "initialized"
    CLOSED = "closed"


class Runner:
    

    

    def __init__(self, *, debug=None, loop_factory=None):
        self._state = _State.CREATED
        self._debug = debug
        self._loop_factory = loop_factory
        self._loop = None
        self._context = None
        self._interrupt_count = 0
        self._set_event_loop = False

    def __enter__(self):
        self._lazy_init()
        return self

    def __exit__(self, exc_type, exc_val, exc_tb):
        self.close()

    def close(self):
        
        if self._state is not _State.INITIALIZED:
            return
        try:
            loop = self._loop
            _cancel_all_tasks(loop)
            loop.run_until_complete(loop.shutdown_asyncgens())
            loop.run_until_complete(
                loop.shutdown_default_executor(constants.THREAD_JOIN_TIMEOUT))
        finally:
            if self._set_event_loop:
                events.set_event_loop(None)
            loop.close()
            self._loop = None
            self._state = _State.CLOSED

    def get_loop(self):
        
        self._lazy_init()
        return self._loop

    def run(self, coro, *, context=None):
        
        if not coroutines.iscoroutine(coro):
            raise ValueError("a coroutine was expected, got {!r}".format(coro))

        if events._get_running_loop() is not None:
            
            raise RuntimeError(
                "Runner.run() cannot be called from a running event loop")

        self._lazy_init()

        if context is None:
            context = self._context
        task = self._loop.create_task(coro, context=context)

        if (threading.current_thread() is threading.main_thread()
            and signal.getsignal(signal.SIGINT) is signal.default_int_handler
        ):
            sigint_handler = functools.partial(self._on_sigint, main_task=task)
            try:
                signal.signal(signal.SIGINT, sigint_handler)
            except ValueError:
                
                
                
                sigint_handler = None
        else:
            sigint_handler = None

        self._interrupt_count = 0
        try:
            return self._loop.run_until_complete(task)
        except exceptions.CancelledError:
            if self._interrupt_count > 0:
                uncancel = getattr(task, "uncancel", None)
                if uncancel is not None and uncancel() == 0:
                    raise KeyboardInterrupt()
            raise  
        finally:
            if (sigint_handler is not None
                and signal.getsignal(signal.SIGINT) is sigint_handler
            ):
                signal.signal(signal.SIGINT, signal.default_int_handler)

    def _lazy_init(self):
        if self._state is _State.CLOSED:
            raise RuntimeError("Runner is closed")
        if self._state is _State.INITIALIZED:
            return
        if self._loop_factory is None:
            self._loop = events.new_event_loop()
            if not self._set_event_loop:
                
                
                events.set_event_loop(self._loop)
                self._set_event_loop = True
        else:
            self._loop = self._loop_factory()
        if self._debug is not None:
            self._loop.set_debug(self._debug)
        self._context = contextvars.copy_context()
        self._state = _State.INITIALIZED

    def _on_sigint(self, signum, frame, main_task):
        self._interrupt_count += 1
        if self._interrupt_count == 1 and not main_task.done():
            main_task.cancel()
            
            self._loop.call_soon_threadsafe(lambda: None)
            return
        raise KeyboardInterrupt()


def run(main, *, debug=None, loop_factory=None):
    
    if events._get_running_loop() is not None:
        
        raise RuntimeError(
            "asyncio.run() cannot be called from a running event loop")

    with Runner(debug=debug, loop_factory=loop_factory) as runner:
        return runner.run(main)


def _cancel_all_tasks(loop):
    to_cancel = tasks.all_tasks(loop)
    if not to_cancel:
        return

    for task in to_cancel:
        task.cancel()

    loop.run_until_complete(tasks.gather(*to_cancel, return_exceptions=True))

    for task in to_cancel:
        if task.cancelled():
            continue
        if task.exception() is not None:
            loop.call_exception_handler({
                'message': 'unhandled exception during asyncio.run() shutdown',
                'exception': task.exception(),
                'task': task,
            })