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
– dirn, Commented Sep 28, 2022 at 16:30

cipres · Accepted Answer · 2022-09-28 17:11:44Z

10

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, 2022 at 17:11

answered Sep 28, 2022 at 16:17

cipres

1965 bronze badges

Sign up to request clarification or add additional context in comments.

dirn Over a year ago

You can shorten the version check to sys.version_info < (3, 10).

Paul Cornelius Over a year ago

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.

Collectives™ on Stack Overflow

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

1 Answer 1

Your Answer

Hot Network Questions

Collectives™ on Stack Overflow

1 Answer 1

2 Comments

Your Answer

Sign up or log in

Post as a guest