如何实现Python通用等待函数：接收目标函数并等待至超时

完全可以实现一个通用的等待函数，核心思路是让它接收一个判断条件的可调用对象（返回布尔值），循环检查直到条件满足或超时。

通用等待函数代码 #

import time

def generic_wait(timeout: int, check_func, sleep_interval: int = 2):
    start_time = time.time()
    while time.time() - start_time < timeout:
        if check_func():
            return
        time.sleep(sleep_interval)
    raise TimeoutError(f"等待超时，已超过{timeout}秒")

针对用户状态场景的用法 #

你原来的期望用法存在一个问题：直接传入wait_until_user_status(user='bil', status='connected')会立即执行该函数，而非将检查逻辑传入通用函数。正确的做法是用lambda表达式封装检查逻辑：

# 假设已定义get_status函数
def get_status(user_name: str) -> str:
    # 数据库连接逻辑
    user = ...
    return user.status

# 调用通用等待函数
try:
    generic_wait(
        timeout=120,
        check_func=lambda: get_status('bil') == 'connected'
    )
    print("用户状态已变为connected")
except TimeoutError as e:
    print(e)

如果想让逻辑更清晰，也可以先单独定义检查函数：

def is_user_connected(user_name: str):
    return get_status(user_name) == 'connected'

try:
    generic_wait(
        timeout=120,
        check_func=lambda: is_user_connected('bil')
    )
except TimeoutError as e:
    print(e)

其他场景示例 #

这个通用函数可以复用在任何需要等待状态满足的场景，比如等待文件存在：

import os

def file_exists(file_path: str):
    return os.path.exists(file_path)

try:
    generic_wait(
        timeout=60,
        check_func=lambda: file_exists('/tmp/data.txt'),
        sleep_interval=1  # 自定义检查间隔
    )
    print("目标文件已存在")
except TimeoutError as e:
    print(e)

参数说明 #

• timeout: 最大等待时长（单位：秒）
• check_func: 无参数的可调用对象，返回布尔值，返回True时停止等待
• sleep_interval: 两次检查之间的休眠时长，默认2秒

内容的提问来源于stack exchange，提问作者Ran Pol