如何实现通用Waiter函数：满足条件后返回首个输入值

首先帮你解决函数名称的困惑：这类反复轮询直到条件满足并返回结果的函数，通常被称为轮询函数（polling function）或者等待值函数（wait-for-value function），搜这个关键词就能找到更多相关资料啦。

接下来针对你的需求，我们可以重构原有的wait_until函数，让它清晰分离"生成值"和"判断条件"的逻辑，同时满足每次重新计算值、返回符合条件结果的要求。

推荐实现方式（安全且灵活） #

这种方案用可调用对象作为参数，避开了字符串eval的安全风险，也更贴合Python的编程习惯：

import time
import logging

logging.basicConfig(level=logging.DEBUG)
log = logging.getLogger(__name__)

def wait_until(producer, predicate, timeout=30, period=1) -> any:
    """等待直到producer生成的值满足predicate条件，然后返回该值"""
    end_time = time.time() + timeout
    while time.time() < end_time:
        try:
            current_value = producer()
            if predicate(current_value):
                log.debug(f"找到符合条件的值: {current_value}")
                return current_value
            else:
                log.debug(f"值 {current_value} 不满足条件")
        except Exception as e:
            log.debug(f"生成值时捕获异常: {e}")
        time.sleep(period)
    log.error("等待超时")
    raise TimeoutError(f"等待{timeout}秒后仍未找到符合条件的值")

使用示例 #

from random import randint

# 找到第一个偶数并返回
result = wait_until(
    producer=lambda: randint(0, 9),
    predicate=lambda x: x % 2 == 0
)
print(result)  # 输出类似4、6这样的偶数

# 尝试找等于10的数（会超时）
try:
    wait_until(
        producer=lambda: randint(0, 9),
        predicate=lambda x: x == 10,
        timeout=5
    )
except TimeoutError as e:
    print(e)  # 输出: 等待5秒后仍未找到符合条件的值

支持字符串条件的版本（谨慎使用） #

如果你确实想用字符串作为判断条件，我们可以用eval实现，但要特别注意：eval会执行任意代码，如果字符串来自不可信来源，会有严重安全风险，仅在你完全控制输入内容的场景下使用：

import time
import logging

logging.basicConfig(level=logging.DEBUG)
log = logging.getLogger(__name__)

def wait_until(producer, condition_str, timeout=30, period=1) -> any:
    """等待直到producer生成的值满足字符串条件，然后返回该值"""
    end_time = time.time() + timeout
    while time.time() < end_time:
        try:
            current_value = producer()
            # 将生成的值绑定到变量x，用eval执行条件判断
            if eval(condition_str, {}, {"x": current_value}):
                log.debug(f"找到符合条件的值: {current_value}")
                return current_value
            else:
                log.debug(f"值 {current_value} 不满足条件: {condition_str}")
        except Exception as e:
            log.debug(f"捕获异常: {e}")
        time.sleep(period)
    log.error("等待超时")
    raise TimeoutError(f"input expression was not \"{condition_str}\" in less than {timeout} seconds")

使用示例 #

from random import randint

# 找到第一个偶数并返回
result = wait_until(lambda: randint(0, 9), "x % 2 == 0")
print(result)  # 输出类似4

# 尝试找等于10的数（会超时）
try:
    wait_until(lambda: randint(0, 9), "x == 10", timeout=5)
except TimeoutError as e:
    print(e)  # 输出: input expression was not "x == 10" in less than 5 seconds

关键改动说明 #

1. 拆分逻辑：把原来的单个expression拆成producer（生成值的函数）和判断条件，确保每次循环都会重新生成新值，满足你"值会变化"的需求。
2. 返回结果：条件满足时直接返回当前生成的值，不再返回None。
3. 异常处理：保留了原有的异常捕获逻辑，避免单次生成值时的小错误导致整个函数崩溃。
4. 提示优化：自定义了超时错误的提示信息，让报错内容更清晰直观。

内容的提问来源于stack exchange，提问作者Todd Bradley