有没有办法使用 atexit 运行 asyncio 协程？

Question

我正在使用`discord.py 开发一个机器人。机器人创建/删除多个频道，并连接到 SQLite 数据库。如果机器人崩溃，我希望它崩溃

1. 销毁它创建的所有临时语音通道。
2. 断开与 SQL 数据库的连接。

这是关闭协程：

async def shutdown(self):
    print("Shutting down Canvas...")
    for ch in self.active_channels:
        await client.delete_channel(ch)
    self.db.close()

我尝试过的事情：

# Canv is the interface between the bot and the data we're collecting
atexit.register(canv.shutdown) 
bot.run(TOKEN)

---

try:
    bot.loop.run_until_complete(bot.start(TOKEN))
except KeyboardInterrupt or InterruptedError:
    bot.loop.run_until_complete(canv.shutdown())
finally:
    bot.loop.close()

---

from async_generator import asynccontextmanager

@asynccontextmanager
async def cleanup_context_manager():
    try:
        yield
    finally:
        await canv.shutdown()

with cleanup_context_manager():
    bot.run(TOKEN)

这些都没有运行canv.shutdown()，这是一个asyncio.coroutine. 如何确保此代码在每种类型的出口上运行？

我用这篇文章来获取一些信息，我认为它最接近我想要的。

Answer 1

你可以尝试这样的事情

import asyncio
import atexit


@atexit.register
def shutdown(self):
    print("Shutting down Canvas...")
    loop = asyncio.get_event_loop()
    loop.run_until_complete(await_delete_channels())
    self.db.close()


async def await_delete_channels(self):
    # # Works but it is better to do it asynchronously
    # for ch in self.active_channels:
    #    await client.delete_channel(ch)
    #
    # Doing delete_channels() asynchronously
    delete_channels = [client.delete_channel(ch) for ch in self.active_channels]
    await asyncio.wait(delete_channels, return_when=asyncio.ALL_COMPLETED)

Answer 2

尝试：

try:
    bot.loop.run_until_complete(bot.start(TOKEN))
finally:
    bot.loop.run_until_complete(canv.shutdown())
    bot.loop.close()

您希望在每种脚本关闭时删除通道并关闭数据库，而不仅仅是在崩溃时，对吧？

否则，请尝试：

try:
    bot.loop.run_until_complete(bot.start(TOKEN))
except Exception:
    bot.loop.run_until_complete(canv.shutdown())
    raise
finally:
    bot.loop.close()

更新：

根据您提供的链接：

引发的异常的信息和异常本身可以通过对 sys.exc_info() 的标准调用来检索。

让我们尝试一下：

import sys

try:
    bot.loop.run_until_complete(bot.start(TOKEN))
finally:
    if sys.exc_info() != (None, None, None):
        bot.loop.run_until_complete(canv.shutdown())
    bot.loop.close()