Python 데코레이터+컨텍스트로 리트라이·타임아웃

from __future__ import annotations

import time
from contextlib import contextmanager
from dataclasses import dataclass


class DeadlineExceeded(TimeoutError):
    pass


@dataclass(frozen=True)
class Deadline:
    end: float

    @staticmethod
    def after(seconds: float) -> "Deadline":
        return Deadline(end=time.monotonic() + seconds)

    def remaining(self) -> float:
        return self.end - time.monotonic()

    def ensure(self) -> None:
        if self.remaining() <= 0:
            raise DeadlineExceeded("deadline exceeded")


@contextmanager
def deadline(timeout: float):
    d = Deadline.after(timeout)
    try:
        yield d
    finally:
        # 여기서 로깅/트레이싱 종료 같은 후처리를 넣기 좋습니다.
        pass

from __future__ import annotations

import functools
import random
import time
from dataclasses import dataclass
from typing import Callable, Iterable, Optional, Type


@dataclass(frozen=True)
class RetryPolicy:
    max_attempts: int = 5
    base_delay: float = 0.2
    max_delay: float = 3.0
    jitter: float = 0.2  # 0.2면 +-20% 랜덤
    retry_on: tuple[Type[BaseException], ...] = (OSError, TimeoutError)


def _sleep_with_deadline(seconds: float, d) -> None:
    # d는 Deadline 또는 None
    if d is None:
        time.sleep(seconds)
        return

    remaining = d.remaining()
    if remaining <= 0:
        raise DeadlineExceeded("deadline exceeded before sleep")

    time.sleep(min(seconds, remaining))


def retry(policy: RetryPolicy, *, deadline_arg: str = "_deadline"):
    """
    `deadline_arg` 이름으로 Deadline을 키워드 인자로 받을 수 있게 해 둡니다.
    호출부에서 전체 타임아웃 컨텍스트와 조합할 때 유용합니다.
    """

    def deco(fn: Callable):
        @functools.wraps(fn)
        def wrapper(*args, **kwargs):
            d = kwargs.get(deadline_arg)

            last_exc: Optional[BaseException] = None
            for attempt in range(1, policy.max_attempts + 1):
                if d is not None:
                    d.ensure()

                try:
                    return fn(*args, **kwargs)
                except policy.retry_on as e:
                    last_exc = e

                    if attempt >= policy.max_attempts:
                        raise

                    # 지수 백오프
                    delay = min(policy.base_delay * (2 ** (attempt - 1)), policy.max_delay)

                    # 지터 적용: delay * (1 - jitter) ~ delay * (1 + jitter)
                    if policy.jitter > 0:
                        factor = 1 + random.uniform(-policy.jitter, policy.jitter)
                        delay = max(0.0, delay * factor)

                    _sleep_with_deadline(delay, d)

            # 논리적으로 도달하지 않지만 타입 안정성을 위해
            if last_exc is not None:
                raise last_exc
            raise RuntimeError("retry wrapper: unreachable")

        return wrapper

    return deco

import requests


policy = RetryPolicy(
    max_attempts=5,
    base_delay=0.3,
    max_delay=2.0,
    jitter=0.3,
    retry_on=(requests.Timeout, requests.ConnectionError),
)


@retry(policy)
def fetch_json(url: str, *, _deadline=None) -> dict:
    # 남은 시간 중 일부를 시도 1회 timeout으로 사용
    per_attempt_timeout = 3.0
    if _deadline is not None:
        remaining = _deadline.remaining()
        _deadline.ensure()
        per_attempt_timeout = max(0.1, min(per_attempt_timeout, remaining))

    r = requests.get(url, timeout=per_attempt_timeout)

    # HTTP 상태코드 기반 재시도는 예외로 변환해서 retry_on에 포함시키는 방식이 깔끔합니다.
    if r.status_code in (429, 500, 502, 503, 504):
        raise requests.Timeout(f"retryable status: {r.status_code}")

    r.raise_for_status()
    return r.json()


def main():
    url = "https://httpbin.org/json"

    with deadline(8.0) as d:
        data = fetch_json(url, _deadline=d)
        print(data)


if __name__ == "__main__":
    main()

import time
from contextlib import contextmanager


@contextmanager
def timed(name: str, logger=print):
    start = time.monotonic()
    try:
        yield
    finally:
        elapsed = time.monotonic() - start
        logger(f"{name} took {elapsed:.3f}s")

with timed("fetch profile"):
    with deadline(5.0) as d:
        profile = fetch_json("https://httpbin.org/json", _deadline=d)