Python 데코레이터로 async 타임아웃·재시도 공통화

import asyncio
import random
import time
from functools import wraps
from typing import Callable, Coroutine, Optional, Tuple, Type


def async_timeout_retry(
    *,
    timeout_s: float,
    max_attempts: int = 3,
    base_delay_s: float = 0.2,
    max_delay_s: float = 3.0,
    jitter_s: float = 0.1,
    retry_on: Tuple[Type[BaseException], ...] = (TimeoutError, ConnectionError, OSError),
    giveup_on: Tuple[Type[BaseException], ...] = (),
    on_retry: Optional[Callable[[int, BaseException, float], None]] = None,
):
    """async 함수에 타임아웃 및 재시도 정책을 적용하는 데코레이터.

    on_retry(attempt, exc, next_delay_s): 재시도 직전에 호출.
    """

    def decorator(func: Callable[..., Coroutine]):
        @wraps(func)
        async def wrapper(*args, **kwargs):
            last_exc: Optional[BaseException] = None
            start = time.perf_counter()

            for attempt in range(1, max_attempts + 1):
                try:
                    # 각 시도마다 타임아웃을 적용
                    return await asyncio.wait_for(func(*args, **kwargs), timeout=timeout_s)

                except asyncio.CancelledError:
                    # 취소는 정책 대상이 아니라 제어 흐름이므로 즉시 전파
                    raise

                except giveup_on as exc:
                    # 즉시 포기할 예외
                    raise

                except retry_on as exc:
                    last_exc = exc
                    if attempt >= max_attempts:
                        break

                    # 지수 백오프: base * 2^(attempt-1)
                    delay = min(max_delay_s, base_delay_s * (2 ** (attempt - 1)))
                    # 지터: 0..jitter_s 랜덤 추가
                    delay = delay + random.random() * jitter_s

                    if on_retry is not None:
                        on_retry(attempt, exc, delay)

                    await asyncio.sleep(delay)

                except Exception as exc:
                    # 기본은 "재시도하지 않는다". 필요하면 retry_on에 추가.
                    raise

            elapsed = time.perf_counter() - start
            raise RuntimeError(
                f"Retry exhausted after {max_attempts} attempts in {elapsed:.3f}s"
            ) from last_exc

        return wrapper

    return decorator

import aiohttp


def log_retry(attempt: int, exc: BaseException, delay: float) -> None:
    print(f"attempt={attempt} failed: {exc!r}, retry in {delay:.2f}s")


@async_timeout_retry(
    timeout_s=2.0,
    max_attempts=4,
    retry_on=(asyncio.TimeoutError, aiohttp.ClientError, OSError),
    on_retry=log_retry,
)
async def fetch_json(session: aiohttp.ClientSession, url: str) -> dict:
    async with session.get(url) as resp:
        resp.raise_for_status()
        return await resp.json()


async def main():
    async with aiohttp.ClientSession() as session:
        data = await fetch_json(session, "https://api.example.com/v1/items")
        print(data)

from typing import Any


def async_timeout_retry_if(
    *,
    timeout_s: float,
    max_attempts: int = 3,
    should_retry: Callable[[BaseException], bool],
    base_delay_s: float = 0.2,
    max_delay_s: float = 3.0,
):
    def decorator(func):
        @wraps(func)
        async def wrapper(*args, **kwargs):
            last_exc = None
            for attempt in range(1, max_attempts + 1):
                try:
                    return await asyncio.wait_for(func(*args, **kwargs), timeout=timeout_s)
                except asyncio.CancelledError:
                    raise
                except Exception as exc:
                    if not should_retry(exc) or attempt >= max_attempts:
                        raise
                    last_exc = exc
                    delay = min(max_delay_s, base_delay_s * (2 ** (attempt - 1)))
                    await asyncio.sleep(delay)
            raise RuntimeError("Retry exhausted") from last_exc

        return wrapper

    return decorator

import aiohttp


def should_retry_http(exc: BaseException) -> bool:
    if isinstance(exc, asyncio.TimeoutError):
        return True
    if isinstance(exc, aiohttp.ClientConnectorError):
        return True
    if isinstance(exc, aiohttp.ClientResponseError):
        return exc.status in (429, 500, 502, 503, 504)
    return False


@async_timeout_retry_if(timeout_s=2.0, max_attempts=5, should_retry=should_retry_http)
async def fetch_text(session: aiohttp.ClientSession, url: str) -> str:
    async with session.get(url) as resp:
        resp.raise_for_status()
        return await resp.text()

import asyncio
import time


def with_deadline(deadline_s: float):
    def decorator(func):
        @wraps(func)
        async def wrapper(*args, **kwargs):
            end = time.perf_counter() + deadline_s
            while True:
                remaining = end - time.perf_counter()
                if remaining <= 0:
                    raise TimeoutError("overall deadline exceeded")
                return await asyncio.wait_for(func(*args, **kwargs), timeout=remaining)
        return wrapper
    return decorator