Python 데코레이터로 async 타임아웃·재시도 패턴

import asyncio
import functools
from typing import Any, Awaitable, Callable, TypeVar

T = TypeVar("T")


def with_timeout(seconds: float):
    def decorator(fn: Callable[..., Awaitable[T]]):
        @functools.wraps(fn)
        async def wrapper(*args: Any, **kwargs: Any) -> T:
            # Python 3.11+
            try:
                async with asyncio.timeout(seconds):
                    return await fn(*args, **kwargs)
            except TimeoutError as e:
                # TimeoutError는 asyncio.timeout이 던지는 표준 예외
                raise

        return wrapper

    return decorator


@with_timeout(1.5)
async def fetch_user(user_id: str) -> dict:
    await asyncio.sleep(2)
    return {"id": user_id}

import asyncio
import functools
import random
import time
from dataclasses import dataclass
from typing import Any, Awaitable, Callable, Optional, Sequence, Type, TypeVar

T = TypeVar("T")


@dataclass(frozen=True)
class RetryState:
    attempt: int
    max_attempts: int
    timeout_seconds: float
    sleep_seconds: float
    elapsed_seconds: float
    exc: BaseException


def async_timeout_retry(
    *,
    timeout_seconds: float,
    max_attempts: int = 3,
    base_delay: float = 0.2,
    max_delay: float = 2.0,
    retry_exceptions: Sequence[Type[BaseException]] = (TimeoutError, OSError),
    on_retry: Optional[Callable[[RetryState], None]] = None,
):
    if max_attempts < 1:
        raise ValueError("max_attempts must be >= 1")

    def decorator(fn: Callable[..., Awaitable[T]]):
        @functools.wraps(fn)
        async def wrapper(*args: Any, **kwargs: Any) -> T:
            start = time.monotonic()
            last_exc: Optional[BaseException] = None

            for attempt in range(1, max_attempts + 1):
                try:
                    async with asyncio.timeout(timeout_seconds):
                        return await fn(*args, **kwargs)
                except tuple(retry_exceptions) as e:
                    last_exc = e
                    if attempt >= max_attempts:
                        break

                    # exponential backoff with full jitter
                    cap = min(max_delay, base_delay * (2 ** (attempt - 1)))
                    sleep_seconds = random.uniform(0, cap)

                    state = RetryState(
                        attempt=attempt,
                        max_attempts=max_attempts,
                        timeout_seconds=timeout_seconds,
                        sleep_seconds=sleep_seconds,
                        elapsed_seconds=time.monotonic() - start,
                        exc=e,
                    )
                    if on_retry:
                        on_retry(state)

                    await asyncio.sleep(sleep_seconds)

            assert last_exc is not None
            raise last_exc

        return wrapper

    return decorator

import logging

logger = logging.getLogger(__name__)


def log_retry(state):
    logger.warning(
        "retrying attempt=%s/%s timeout=%.2fs sleep=%.2fs elapsed=%.2fs exc=%r",
        state.attempt,
        state.max_attempts,
        state.timeout_seconds,
        state.sleep_seconds,
        state.elapsed_seconds,
        state.exc,
    )


@async_timeout_retry(timeout_seconds=0.8, max_attempts=4, on_retry=log_retry)
async def call_dependency() -> str:
    # 외부 호출이라고 가정
    await asyncio.sleep(2)
    return "ok"

from typing import Protocol


class RetryIf(Protocol):
    def __call__(self, exc: BaseException) -> bool: ...


def async_timeout_retry2(
    *,
    timeout_seconds: float,
    max_attempts: int = 3,
    base_delay: float = 0.2,
    max_delay: float = 2.0,
    retry_exceptions=(TimeoutError, OSError),
    retry_if: RetryIf = lambda exc: True,
):
    def decorator(fn):
        @functools.wraps(fn)
        async def wrapper(*args, **kwargs):
            last_exc = None
            for attempt in range(1, max_attempts + 1):
                try:
                    async with asyncio.timeout(timeout_seconds):
                        return await fn(*args, **kwargs)
                except tuple(retry_exceptions) as e:
                    if not retry_if(e):
                        raise
                    last_exc = e
                    if attempt == max_attempts:
                        break
                    cap = min(max_delay, base_delay * (2 ** (attempt - 1)))
                    await asyncio.sleep(random.uniform(0, cap))
            raise last_exc

        return wrapper

    return decorator

import asyncio
import pytest


@pytest.mark.asyncio
async def test_retry_eventually_success():
    counter = {"n": 0}

    @async_timeout_retry(timeout_seconds=0.05, max_attempts=3, base_delay=0.0, max_delay=0.0)
    async def flaky():
        counter["n"] += 1
        if counter["n"] < 2:
            raise OSError("temporary")
        return "ok"

    assert await flaky() == "ok"
    assert counter["n"] == 2