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?
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.