重试策略

重试策略 #

基类：Protocol

用于控制失败后步骤重试的策略接口。

实现方案根据已用时间、尝试次数和最后错误来决定是否重试以及下次尝试前的等待时长。

相关链接

ConstantDelayRetryPolicy
步骤

workflows/retry_policy.py中的源代码

@runtime_checkable
class RetryPolicy(Protocol):
    """
    Policy interface to control step retries after failures.

    Implementations decide whether to retry and how long to wait before the next
    attempt based on elapsed time, number of attempts, and the last error.

    See Also:
        - [ConstantDelayRetryPolicy][workflows.retry_policy.ConstantDelayRetryPolicy]
        - [step][workflows.decorators.step]
    """

    def next(
        self, elapsed_time: float, attempts: int, error: Exception
    ) -> float | None:
        """
        Decide if another retry should occur and the delay before it.

        Args:
            elapsed_time (float): Seconds since the first failure.
            attempts (int): Number of attempts made so far.
            error (Exception): The last exception encountered.

        Returns:
            float | None: Seconds to wait before retrying, or `None` to stop.
        """

下一页 #

next(elapsed_time: float, attempts: int, error: Exception) -> float | None

决定是否应进行另一次重试以及重试前的延迟时间。

参数：

名称	类型	描述	默认
`elapsed_time`	`float`	自首次故障以来的秒数。	必填
`attempts`	`int`	截至目前已尝试的次数。	必填
`error`	`Exception`	遇到的最后一个异常。	必填

返回：

类型	描述
`float \| None`	float \| None: 重试前等待的秒数，或 `None` 表示停止。

workflows/retry_policy.py中的源代码

def next(
    self, elapsed_time: float, attempts: int, error: Exception
) -> float | None:
    """
    Decide if another retry should occur and the delay before it.

    Args:
        elapsed_time (float): Seconds since the first failure.
        attempts (int): Number of attempts made so far.
        error (Exception): The last exception encountered.

    Returns:
        float | None: Seconds to wait before retrying, or `None` to stop.
    """

恒定延迟重试策略 #

以固定间隔重试，最多尝试指定次数。

示例：

@step(retry_policy=ConstantDelayRetryPolicy(delay=5, maximum_attempts=10))
async def flaky(self, ev: StartEvent) -> StopEvent:
    ...

workflows/retry_policy.py中的源代码

class ConstantDelayRetryPolicy:
    """Retry at a fixed interval up to a maximum number of attempts.

    Examples:
        ```python
        @step(retry_policy=ConstantDelayRetryPolicy(delay=5, maximum_attempts=10))
        async def flaky(self, ev: StartEvent) -> StopEvent:
            ...
        ```
    """

    def __init__(self, maximum_attempts: int = 3, delay: float = 5) -> None:
        """
        Initialize the policy.

        Args:
            maximum_attempts (int): Maximum consecutive attempts. Defaults to 3.
            delay (float): Seconds to wait between attempts. Defaults to 5.
        """
        self.maximum_attempts = maximum_attempts
        self.delay = delay

    def next(
        self, elapsed_time: float, attempts: int, error: Exception
    ) -> float | None:
        """Return the fixed delay while attempts remain; otherwise `None`."""
        if attempts >= self.maximum_attempts:
            return None

        return self.delay

下一页 #

next(elapsed_time: float, attempts: int, error: Exception) -> float | None

在尝试次数仍剩余时返回固定延迟；否则 None。

workflows/retry_policy.py中的源代码

def next(
    self, elapsed_time: float, attempts: int, error: Exception
) -> float | None:
    """Return the fixed delay while attempts remain; otherwise `None`."""
    if attempts >= self.maximum_attempts:
        return None

    return self.delay