loop_run

privex.helpers.asyncx.loop_run(coro: Union[Coroutine, Type[Coroutine], Callable], *args, _loop=None, **kwargs) → Any[source]

Run the coroutine or async function coro synchronously, using an AsyncIO event loop.

If the keyword argument _loop isn’t specified, it defaults to the loop returned by asyncio.get_event_loop()

If coro doesn’t appear to be a coroutine or async function:

  • If coro is a normal callable object e.g. a function, then it’ll be called.

    • If the object returned after calling coro(*args, **kwargs) is a co-routine / async func, then it’ll call loop_run again, passing the object returned from calling it, and returning the result from that recursive call.

    • If the returned object isn’t an async func / co-routine, then the object will be returned as-is.

  • Otherwise, coro will just be returned back to the caller.

Example Usage

First we’ll define the async function some_func to use as an example:

>>> async def some_func(x, y):
...     return x + y

Option 1 - Call an async function directly with any args/kwargs required, then pass the coroutine returned:

>>> loop_run(some_func(3, 4))
7

Option 2 - Pass a reference to the async function, and pass any required args/kwargs straight to loop_run() - the function will be ran with the args/kwargs you provide, then the coroutine ran in an event loop:

>>> loop_run(some_func, 10, y=20)    # Opt 2. Pass the async function and include any args/kwargs for the call
30
Parameters
  • coro – A co-routine, or reference to an async function to be ran synchronously

  • args – Any positional arguments to pass to coro (if it’s a function reference and not a coroutine)

  • _loop (asyncio.base_events.BaseEventLoop) – (kwarg only!) If passed, will run coro in this event loop, instead of asyncio.get_event_loop()

  • kwargs – Any keyword arguments to pass to coro (if it’s a function reference and not a coroutine)

Return Any coro_result

The returned data from executing the coroutine / async function