How to replace asyncio.get_event_loop() to avoid the DeprecationWarning?

Question

The code below works, but I'm a bit confused by the:

DeprecationWarning: There is no current event loop
  loop = asyncio.get_event_loop()

and the

DeprecationWarning: There is no current event loop
  loop.run_until_complete(asyncio.gather(*coroutine_list))

warnings.

I've found this and this but what I'm confused about is how to update my code. I understand that there isn't an even loop yet, but how do I set it? I assumed that I needed to change asyncio.get_event_loop() with asyncio.new_event_loop() but this doesn't seem to work.

This is my code

async def flush(self, queue_c) -> None:
    try:
        loop = asyncio.get_running_loop()
        result_objects_action = loop.run_in_executor(None, self.create_objects, queue_c)
        # do stuff
    except:
        # except handling

def _run_queue(self) -> None:
    coroutine_list = []
    c = 0
    while c < len(self._queue):
        coroutine_list.append(self.flush(c))
        c+=1
    # What should be changed here?
    loop = asyncio.get_event_loop()
    loop.run_until_complete(asyncio.gather(*coroutine_list))
    self._queue = []

The preferred way to start a new loop in `asyncio.run`, though you can't pass `asyncio.gather` to it directly, you'd need to create a separate coroutine to wrap it. `async def main` is a common pattern for that. — dirn, Sep 28 '22 at 16:30

cipres · Answer 1 · 2022-09-28T17:11:44.680

4

Calling asyncio.new_event_loop() isn't enough, you need to set it after with asyncio.set_event_loop(loop). This deprecation warning for get_event_loop() was introduced by python 3.10 i believe. Here's how i setup the event loop so that it works with any python 3.x version:

import sys
import asyncio

if sys.version_info < (3, 10):
    loop = asyncio.get_event_loop()
else:
    try:
        loop = asyncio.get_running_loop()
    except RuntimeError:
        loop = asyncio.new_event_loop()

    asyncio.set_event_loop(loop)

edited Sep 28 '22 at 17:11

answered Sep 28 '22 at 16:17

cipres

106
3

You can shorten the version check to `sys.version_info < (3, 10)`. – dirn Sep 28 '22 at 16:29
Why can't you just do `loop = asyncio.new_event_loop; asyncio.set_event_loop(loop)`? That will work with all versions, I think. You would only do this if, for some reason, you can't use the simpler asyncio.run(coro) approach. – Paul Cornelius Sep 29 '22 at 00:09

How to replace asyncio.get_event_loop() to avoid the DeprecationWarning?

1 Answers1