使用Python PyInquirer和asyncio创建用户界面以触发异步操作

2024-06-28 19:02:47 发布

您现在位置:Python中文网/ 问答频道 /正文

我想使用PyInquirer和asyncio创建一个用户界面,用于触发异步操作

应该出现一个提示(使用PyInquirer),用户可以从中选择操作。这将创建一个挂起的任务,并再次出现提示。当挂起的任务完成时,结果将打印到标准输出,即使提示仍在屏幕上

如果我只想让它运行一次,这不是问题,可以这样做:

import asyncio
from PyInquirer import prompt

options = {
'type': 'list',
'name': 'functionToCall',
'message': 'Choose Command',
'choices': [
    'Say Hello',
    'Exit'
]
}

async def sayHello():
    await asyncio.sleep(2)
    print("hello")

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

async def main():
    answers = prompt(options)
    kill = False;
    if answers['functionToCall'] == 'Say Hello':
        loop.create_task(sayHello())
    else:
        kill = True
        loop.stop()

if __name__ == "__main__":
    loop.create_task(main())
    loop.run_forever()

如果选择“打招呼”,它将在2秒后打印到输出

但是,如果我希望在选择上一个操作后立即显示提示,这将阻止结果出现。即

async def main():
    answers = prompt(options)
    kill = False;
    if answers['functionToCall'] == 'Say Hello':
        loop.create_task(sayHello())
    else:
        kill = True
        loop.stop()

    if not kill:
        loop.create_task(main())

选择最后一个操作后(根据需要),提示将立即出现,但“Hello”永远不会打印到输出中。我认为这是因为PyInquire中的prompt()函数阻止了其他任务的执行

当程序在blockingprompt()函数中时,如何让sayHello()任务在后台执行,然后让sayHello()命令打印它的结果,即使仍然在prompt()函数中

谢谢


Tags: loopasynciohellotaskifmaincreateprompt
1条回答
网友
1楼 · 发布于 2024-06-28 19:02:47

我已经找到了答案:

可以告诉prompt函数patch_stdout = Truereturn_asyncio_coroutine = True

这将使它返回未来字符串的dict,而不是字符串的dict。 以后的字符串可以作为异步任务处理

import asyncio
from PyInquirer import prompt

options = {
'type': 'list',
'name': 'functionToCall',
'message': 'Choose Command',
'choices': [
    'Say Hello',
    'Exit'
]
}

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


async def sayHello():
    await asyncio.sleep(2)
    print("hello!")

def handleChoice(future):
    choice = future.result()
    if choice == "Say Hello":
        loop.create_task(sayHello())
        loop.create_task(myprompt())
    else:
        loop.stop()

async def myprompt():
    future_answers_generator = prompt(options, patch_stdout = True, return_asyncio_coroutine = True)['functionToCall']    
    future = asyncio.ensure_future(future_answers_generator)
    future.add_done_callback(handleChoice)


if __name__ == "__main__":
    loop.create_task(myprompt())
    loop.run_forever()

相关问题 更多 >