API 异步任务轮询计算引擎（专业版）

输入 / 输出

输入：job_id、幂等键、初始/当前状态；退避参数（base、factor、max_interval）、
当前轮询次数、任务开始与当前时间、最大等待。
输出：下次轮询间隔、累计等待、是否超时、当前状态与 HTTP 响应（状态码/body）。

使用示例

from async_job import AsyncJob, next_poll_interval, should_timeout, RUNNING, SUCCEEDED

job = AsyncJob(job_id="j1", idem_key="k-123")
job.transition(RUNNING)
# 客户端第 attempt 次轮询前：
wait = next_poll_interval(attempt, base=1, factor=2, max_interval=30)
if should_timeout(started_at, now, max_wait=60):
    ...  # 停止并提示超时
job.transition(SUCCEEDED, result={"url": "..."})
status_code, body, _ = job.http_response()

引擎代码：async_job.py

#!/usr/bin/env python3
# -*- coding: utf-8 -*-
"""API 异步任务轮询计算引擎（专业版核心）
长耗时操作：提交 -> 202 + job_id/status_url -> 客户端指数退避轮询，
状态机 PENDING -> RUNNING -> SUCCEEDED / FAILED；含总超时与重复提交防护。
"""

PENDING, RUNNING, SUCCEEDED, FAILED = "PENDING", "RUNNING", "SUCCEEDED", "FAILED"
TERMINAL = (SUCCEEDED, FAILED)


class AsyncJob:
    def __init__(self, job_id, idem_key=None, status=PENDING, result=None, error=None):
        self.job_id = job_id
        self.idem_key = idem_key
        self.status = status
        self.result = result
        self.error = error

    def transition(self, new_status, result=None, error=None):
        # 合法流转：PENDING->RUNNING->SUCCEEDED/FAILED；终态不可再变
        order = {PENDING: 0, RUNNING: 1, SUCCEEDED: 2, FAILED: 2}
        if self.status in TERMINAL:
            return self.status
        if order.get(new_status, -1) >= order[self.status]:
            self.status = new_status
            if result is not None:
                self.result = result
            if error is not None:
                self.error = error
        return self.status

    def http_response(self):
        """返回 (http_status, body, retry_after)"""
        if self.status in (PENDING, RUNNING):
            return 200 if self.status == RUNNING else 202, {
                "job_id": self.job_id, "status": self.status}, None
        if self.status == SUCCEEDED:
            return 200, {"job_id": self.job_id, "status": SUCCEEDED, "result": self.result}, None
        return 200, {"job_id": self.job_id, "status": FAILED, "error": self.error}, None


def next_poll_interval(attempt, base=1.0, factor=2.0, max_interval=30.0):
    """指数退避：base*factor^attempt，封顶 max_interval；非法入参兜底为 base"""
    if base <= 0 or factor <= 0:
        base = 1.0; factor = 2.0
    interval = base * (factor ** max(attempt, 0))
    # max_interval 为所有轮询间隔的硬上限；配置矛盾(max<base)时以硬上限为准
    if max_interval <= 0:
        max_interval = base
    return min(interval, max_interval)


def poll_schedule(attempts, base=1.0, factor=2.0, max_interval=30.0):
    """生成各次轮询间隔序列"""
    return [next_poll_interval(a, base, factor, max_interval) for a in range(attempts)]


def elapsed_at(attempt, base=1.0, factor=2.0, max_interval=30.0):
    """到第 attempt 次轮询前的累计等待秒数"""
    return sum(poll_schedule(attempt, base, factor, max_interval))


def should_timeout(started_at, now, max_wait):
    """总等待是否超时"""
    return (now - started_at) >= max_wait

测试（3 正例 + 1 边界，实跑全过）

硬性规则

1. 提交快速返回 202，与轮询分离，不做长同步等待。
2. 轮询指数退避、间隔有硬上限、整体有总超时，避免空转。
3. 幂等键防重复任务；终态结果可重复查询、不再变更。
4. 服务端任务需有执行超时与清理，防止永久悬挂。