如何用Python实现带超时的通用状态检查函数？

完全可行，这种重复的超时循环逻辑非常适合封装成通用函数，既减少代码冗余，也方便统一维护超时、轮询这类通用逻辑。

实现思路 #

把每个业务函数的「状态获取逻辑（原函数调用+结果过滤）」单独封装成小函数，然后交给一个通用的超时检查函数去处理循环、超时判断、状态比对：

• 通用函数负责计时、循环轮询、超时返回最后状态
• 业务相关的状态提取逻辑单独实现，作为参数传给通用函数

具体代码实现 #

import time

def wait_for_state(get_current_state, target_state, timeout=15, poll_interval=1):
    """
    通用超时状态检查函数
    :param get_current_state: 回调函数，返回当前需要检查的状态值
    :param target_state: 目标状态值
    :param timeout: 超时时间（秒），默认15秒
    :param poll_interval: 轮询间隔（秒），默认1秒
    :return: 最终状态（达到目标状态则返回目标值，超时则返回最后一次获取的状态）
    """
    start_time = time.time()
    last_state = None
    while time.time() - start_time < timeout:
        last_state = get_current_state()
        if last_state == target_state:
            return last_state
        time.sleep(poll_interval)  # 轮询间隔，避免频繁调用
    print("超时！")
    return last_state

# ---------------------- 业务相关的状态提取函数 ----------------------
def get_function1_state():
    # 调用原函数获取字典列表
    result_list = function_1()
    # 这里替换成你的实际过滤逻辑，比如从列表中提取目标状态
    # 示例：假设取第一个字典的"status"字段
    return result_list[0].get("status") if result_list else None

def get_function2_state():
    result_list = function_2()
    # 示例：遍历字典列表，找到特定id的项并返回其状态
    for item in result_list:
        if item.get("order_id") == "target_order":
            return item.get("status")
    return None

# ---------------------- 调用示例 ----------------------
# 检查function1是否达到"Approved"状态
final_state1 = wait_for_state(get_function1_state, "Approved", timeout=15)
print(f"Function1最终状态: {final_state1}")

# 检查function2是否达到"Delivered"状态
final_state2 = wait_for_state(get_function2_state, "Delivered", timeout=15)
print(f"Function2最终状态: {final_state2}")

优势说明 #

• 彻底消除重复代码：超时判断、循环逻辑只需要维护一次
• 业务逻辑解耦：状态提取和超时检查分开，每个部分职责更清晰
• 灵活性高：可以通过调整poll_interval参数控制轮询频率，适配不同业务场景

内容的提问来源于stack exchange，提问作者Rafa S