如何用Python实现双API调用:触发任务并定时轮询结果状态
问题描述
我有两个API:triggerAPI 和 triggerAPIResult。调用triggerAPI会触发一个耗时数分钟的处理流程;triggerAPIResult用于检查该流程是否完成。当triggerAPIResult返回true时,此时调用triggerAPI才能得到期望输出——流程处理期间triggerAPI返回的是无意义数据。要求每隔1分钟调用一次triggerAPIResult,最多轮询10分钟,请问如何用Python实现?
实现思路与代码
可以通过以下步骤实现这个轮询逻辑:
- 先调用
triggerAPI触发处理流程 - 循环调用
triggerAPIResult检查状态,每次间隔1分钟 - 达到10分钟(即10次轮询)仍未完成则终止,或根据需求处理超时情况
- 一旦
triggerAPIResult返回true,立即调用triggerAPI获取有效结果
下面是完整的Python实现代码:
import time import requests # 假设使用requests库调用API,可根据实际情况替换 def call_trigger_api(): """调用triggerAPI触发处理流程""" # 替换为实际的API请求逻辑,比如添加headers、认证信息等 response = requests.post("https://your-domain/triggerAPI") response.raise_for_status() # 捕获请求异常 print("已触发处理流程") def check_trigger_result(): """调用triggerAPIResult检查处理状态""" # 替换为实际的API请求逻辑 response = requests.get("https://your-domain/triggerAPIResult") response.raise_for_status() return response.json() # 假设接口返回布尔值true/false def get_final_result(): """调用triggerAPI获取最终有效结果""" # 替换为实际的API请求逻辑 response = requests.get("https://your-domain/triggerAPI") response.raise_for_status() return response.json() def main(): # 第一步:触发处理流程 call_trigger_api() max_poll_times = 10 # 最多轮询10次,对应10分钟 poll_interval = 60 # 每次轮询间隔60秒 completed = False for count in range(max_poll_times): print(f"第{count+1}次检查处理状态...") completed = check_trigger_result() if completed: print("处理流程已完成,开始获取有效结果") final_result = get_final_result() print("最终结果:", final_result) break print("处理未完成,等待1分钟后再次检查") time.sleep(poll_interval) else: # 循环正常结束未触发break,说明超时 print("轮询超时(已达10分钟),处理流程未完成") if __name__ == "__main__": main()
关键细节说明
- 用
time.sleep(poll_interval)精准控制轮询间隔,确保每次检查间隔1分钟 - 通过
range(max_poll_times)限制最大轮询次数,刚好对应10分钟的上限 - 加入
response.raise_for_status()处理API请求异常,避免请求失败时程序静默崩溃 - 函数拆分清晰,便于后续修改API调用逻辑(比如更换HTTP库、添加认证、调整请求参数等)
- 超时分支(
else块)可根据实际需求扩展,比如添加告警、记录日志或发起重试
内容的提问来源于stack exchange,提问作者Good Panic
相关产品推荐
相关产品推荐

