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

Question

I'm building an SMTP server with aiosmtpd and used the examples as a base to build from. Below is the code snippet for the entry point to the program.

if __name__ == '__main__':
    loop = asyncio.get_event_loop()
    loop.create_task(amain(loop=loop))
    try:
        loop.run_forever()
    except KeyboardInterrupt:
        pass

When I run the program, I get the following warning:

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

What's the correct way to implement this?

Wow! This change in get_event_loop() behavior went under the radar for me. I am one of the maintainers of aiosmtpd, and we'll try to urgently push a new version to handle this. — pepoluan
– pepoluan, Commented Dec 28, 2022 at 9:05

Paul Cornelius · Accepted Answer · 2023-04-15 08:59:53Z

70

Your code will run on Python3.10 but as of 3.11 it will be an error to call asyncio.get_event_loop when there is no running loop in the current thread. Since you need loop as an argument to amain, apparently, you must explicitly create and set it.

It is better to launch your main task with asyncio.run than loop.run_forever, unless you have a specific reason for doing it that way. [But see below]

Try this:

if __name__ == '__main__':
    loop = asyncio.new_event_loop()
    asyncio.set_event_loop(loop)
    try:
        asyncio.run(amain(loop=loop))
    except KeyboardInterrupt:
        pass

Added April 15, 2023:

There is a difference between calling asyncio.run(), which I have done here, and calling loop.run_forever() (as in the original question) or loop.run_until_complete(). When I wrote this answer I did not realize that asyncio.run() always creates a new event loop. Therefore in my code above, the variable loop that is passed to amain will not become the "running loop." So my code avoids the DeprecationWarning/RuntimeException, but it doesn't pass a useful loop into amain.

To correct that, replace the line

asyncio.run(amain(loop=loop))

with

loop.run_until_complete(amain(loop=loop))

It would be best to modify amain to obtain the running event loop inside the function instead of passing it in. Then you could launch the program with asyncio.run. But if amain cannot be changed that won't be possible.

Note that run_until_complete, unlike asyncio.run, does not clean up async generators. This is documented in the standard docs.

edited Apr 15, 2023 at 8:59

answered Aug 15, 2022 at 23:00

Paul Cornelius

11.4k1 gold badge18 silver badges28 bronze badges

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

10 Comments

darren Over a year ago

do you use amain because it would be different from main ?

Paul Cornelius Over a year ago

I used amain because that was the name used in the original post. Since in the no actual function was provided, the name has no significance. There was no reason to change it.

Jari Turkia Over a year ago

Similar answer @ stackoverflow.com/a/68652643/1548275 suggesting asyncio.new_event_loop() to replace obsoleted get_event_loop().

alelom Over a year ago

As highlighted by the comments in a similar question, this solution may be thread-unsafe. I opted for this solution instead: stackoverflow.com/a/69314701/3873799

darren Over a year ago

the docs suggest this: asyncio.run(main()) . . . here are the docs: docs.python.org/3/library/asyncio-task.html

|

知识来一点 · Accepted Answer · 2024-05-08 01:55:13Z

2

In my case, below code resolve my problem:

loop = asyncio.new_event_loop()
asyncio.set_event_loop(loop)

start_server = websockets.serve(echo, "localhost", 50558)

loop.run_until_complete(start_server)
loop.run_forever()

answered May 8, 2024 at 1:55

知识来一点

274 bronze badges

Comments

seunggabi · Accepted Answer · 2023-08-05 14:48:53Z

-1

async def ws(f):
    pass


async def callback(*args):
    pass


async def create():
    tasks = [
        asyncio.create_task(
            ws(callback)
        )
    ]

    await asyncio.wait(tasks)


if __name__ == "__main__":
    loop = asyncio.new_event_loop()
    asyncio.set_event_loop(loop)

    loop.run_until_complete(create())

answered Aug 5, 2023 at 14:48

seunggabi

1,83016 silver badges13 bronze badges

1 Comment

Melebius Feb 6 at 11:39

What information does this add that is not present in previous answers? Please edit to explain.

Collectives™ on Stack Overflow

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

3 Answers 3

10 Comments

Comments

1 Comment

Your Answer

Linked

Hot Network Questions

Collectives™ on Stack Overflow

3 Answers 3

10 Comments

Comments

1 Comment

Your Answer

Sign up or log in

Post as a guest

Linked

Related