discord.py(带Cogs)+FastAPI:域名解析失败及优雅终止问题
在FastAPI中集成discord.py机器人的问题与后续疑问
初始问题:errno -3 temporary failure in name resolution错误
尝试在FastAPI中启动一个基础discord.py机器人时,遭遇errno -3 temporary failure in name resolution错误,相关代码如下:
@typechecked class UnfilteredBot(Bot): async def process_commands(self, message:Message)->None: ctx:Context = await self.get_context(message) await self.invoke(ctx) @typechecked class MyBot(UnfilteredBot): def __init__(self, command_prefix:str='!')->None: intents: Intents = Intents.default() #intents.members = True # allow to get commands from GC intents.message_content = True # v2 super().__init__(intents=intents, command_prefix=command_prefix) @typechecked def get_app(token:str, guild:str, channel:int, rest_key:str)->FastAPI: shutdown :Event = Event() # from threading import Event #complete :Event = Event() app :FastAPI = FastAPI() bot :Bot = MyBot() @app.on_event("startup") async def startup_event()->Coroutine[None]: # this function will run before the main API starts bot_start:Coroutine[None] = bot.start(token) create_task( bot_start ) @app.on_event("shutdown") async def shutdown_event()->Coroutine[None]: shutdown.set() #while not complete.wait(1): # gently stop a different bot's loop # await sleep(1) # work-around for follow-up question # print("waiting") return app @typechecked def main()->None: load_dotenv () # from dotenv import load_dotenv token :str = environ['DISCORD_TOKEN'] guild :str = environ['DISCORD_GUILD'] channel :int = int(environ['DISCORD_CHANNEL']) host :str = getenv('HOST', 'locahost') # <-- 问题所在 port :int = getenv('PORT', 5000) setup_logging() # from discord.utils import setup_logging app :FastAPI = get_app(token, guild, channel, rest_key) urun(app, host=host, port=port) # from uvicorn import run as urun if __name__ == '__main__': exit(main())
问题解决说明
该错误已解决,原因是host参数拼写错误(locahost应为localhost)。分享此内容是因为目前缺少包含Cogs和优雅终止逻辑的相关示例。
后续疑问:协程环境下的机器人优雅终止问题
当前无法优雅终止另一个机器人,原使用的threading.Event方案仅适用于线程环境,在协程环境下循环停止后finally块无法执行,导致FastAPI服务器无法正常关闭。相关代码如下:
async def start(self)->Coroutine[None]: try: print('AI.start() try') while not self.shutdown.wait(1): print('AI.start() loop') await sleep(1) #??? pass finally: print('AI.start() finally') #self.shutdown.set() # assert is set ? self.complete.set() print('AI.start() done')
已找到一个基于asyncio.Event的Event_ts实现,但原机器人是适配线程应用开发的,不确定该实现是否适用,需要找到协程环境下实现定时wait()的解决方案。
内容的提问来源于stack exchange,提问作者Innovations Anonymous
相关产品推荐
相关产品推荐

