中文
devexamples

Tenacity Retry Stop Wait Strategies

Compose retry behavior with Tenacity using stop conditions, adaptive wait strategies, custom retry predicates, and typed final errors.

PythonResilience

Code

snippet
import random
from typing import Any

from tenacity import (
    RetryError,
    retry,
    stop_after_attempt,
    stop_after_delay,
    wait_random_exponential,
    before_sleep_log,
    retry_if_exception_type,
)
import logging

logging.basicConfig(level=logging.INFO)
logger = logging.getLogger(__name__)


class ThrottleError(Exception):
    pass


@retry(
    stop=(stop_after_attempt(6) | stop_after_delay(30)),
    wait=wait_random_exponential(multiplier=0.5, max=10),
    retry=retry_if_exception_type((ConnectionError, ThrottleError)),
    before_sleep=before_sleep_log(logger, logging.WARNING),
    reraise=True,
)
def fetch_ledger(page: int) -> list[dict[str, Any]]:
    roll = random.random()
    if roll < 0.35:
        raise ConnectionError('ledger node unreachable')
    if roll < 0.5:
        raise ThrottleError('quota exceeded for this window')
    return [{'page': page, 'rows': random.randint(1, 40)}]


def safe_fetch(page: int) -> list[dict[str, Any]]:
    try:
        return fetch_ledger(page)
    except RetryError:
        return []
    except (ConnectionError, ThrottleError):
        return []  # reraise=True 时最终以原始异常抛出


rows = safe_fetch(1)
print('ledger rows fetched:', len(rows))

Related Examples