【发布时间】:2023-01-26 20:44:51
【问题描述】:
所以我有这个函数可以从数据库中获取我的用户状态:
def get_status(user_name: str) -> str:
# connect to db.
user = ...
return user.status
现在我必须等待最多 time,直到我的用户 status 变为 connected(可能需要几分钟):
def wait_until_user_status(user: str, timeout: int)
status = get_status(user)
start = time.time()
while time.time() - start < timeout and status != `connected`:
time.sleep(2)
status = get_status(user)
if status != `connected`:
raise Exception(f'Timeout of {timeout} seconds reached')
用法:
try:
wait_for('bil', 120)
except Exception as ex:
print(ex)
所以我的问题是: 如果我有几个函数想验证内部的东西(在上面的例子中 - 用户状态)是否可以声明一个通用函数来接受另一个函数并等到超时? (在我的示例中 - 发送返回状态并等待某个条件的函数)。
例如:
def generic_wait(timeout: int, function):
pass
用法:
generic_wait(timeout=120, function=wait_until_user_status(user='bil', status='connected')
【问题讨论】:
-
我认为您需要将
get_status设为异步函数。 asyncio.timeout -
请查看我的更新,我想要一个通用的解决方案
-
我的观点仍然是:如果提供给
generic_wait的function是同步的,这意味着在等待函数完成时不能做任何其他事情,那么程序将如何检查是否发生超时? -
这次我想让我的功能等待,同时不需要继续
标签: python time delegates wait