基于Python pytest自动化框架实现Job Status API状态轮询与日志输出

问题核心需求 #

• 周期性调用Job Status API查询作业状态
• 实时打印每次查询到的中间状态
• 直到API返回status = "Job is completed successfully"时停止等待
• 禁止硬编码长时间sleep，改用短周期轮询机制

代码修改方案 #

1. 修正exec_request函数参数与异常处理 #

原函数缺少必要参数，且未处理请求异常，先补全并增强鲁棒性：

import requests
import json
import time
import logging

logger = logging.getLogger(__name__)

def exec_request(request_type, api_url, headers, payload):
    try:
        api_response = requests.Session().request(
            method=request_type,
            url=api_url,
            headers=headers,
            params=payload,
            verify=False
        )
        api_response.raise_for_status()  # 捕获HTTP错误状态码
        api_response_content = json.loads(api_response.content.decode('utf-8'))
        return {
            'content': api_response_content,
            'status_code': api_response.status_code
        }
    except Exception as e:
        logger.error(f"API请求失败: {str(e)}")
        return None

2. 封装作业状态获取函数 #

单独封装获取状态的逻辑，简化后续调用：

def get_job_status(job_id, access_token):
    headers = {
        "authorization": access_token
    }
    # 修正URL拼接逻辑（原代码缺少分隔符，会生成错误URL）
    latest_job_status_url = f"http://localhost/{job_id}/latest_status"
    response = exec_request("GET", latest_job_status_url, headers, "null")
    
    if response and response['status_code'] == 200:
        # 与API返回字段对齐，这里取status字段
        return response['content'].get('status')
    
    logger.error("获取作业状态失败")
    return None

3. 实现等待作业完成的核心逻辑 #

修正原尝试代码的循环逻辑错误，添加实时状态打印：

def wait_for_job_completion(job_id, access_token, poll_interval=10):
    """
    等待作业完成，周期性打印中间状态
    :param job_id: 目标作业ID
    :param access_token: 接口认证令牌
    :param poll_interval: 轮询间隔（秒，默认10秒）
    :return: 最终完成状态
    """
    while True:
        current_status = get_job_status(job_id, access_token)
        
        if not current_status:
            # 获取状态失败，等待后重试
            time.sleep(poll_interval)
            continue
        
        # 打印当前中间状态
        logger.info(f"---当前作业状态: {current_status}")
        
        # 判断是否达到完成状态
        if current_status == "Job is completed successfully":
            logger.info("---作业已成功完成")
            return current_status
        
        # 未完成则等待指定间隔后继续轮询
        time.sleep(poll_interval)

4. 调用示例 #

# 替换为实际的认证令牌和作业ID
access_token = "your_valid_access_token"
job_id = "target_job_id"

# 启动等待逻辑
wait_for_job_completion(job_id, access_token)

关键修改说明 #

• 修复了exec_request的参数缺失问题，添加异常捕获避免请求失败导致程序崩溃
• 修正URL拼接错误（原代码缺少分隔符，会生成localhost123/latest_status这类无效URL）
• 调整循环逻辑：原代码的or条件会导致无限循环，改为直接判断目标完成状态
• 每次轮询都打印当前状态，满足实时输出中间状态的需求
• 轮询间隔可配置，避免硬编码固定sleep时长
• 处理了状态获取失败的场景，增加重试机制

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