실무형 Python 데코레이터로 리트라이·캐시·타임아웃 구현

import time
import random
import functools
from typing import Callable, Iterable, Tuple, Type


def retry(
    *,
    retries: int = 3,
    base_delay: float = 0.2,
    max_delay: float = 3.0,
    jitter: float = 0.1,
    retry_on: Tuple[Type[BaseException], ...] = (TimeoutError, ConnectionError),
    on_retry: Callable[[int, BaseException, float], None] | None = None,
):
    """동기 함수용 리트라이 데코레이터.

    - retries: 총 시도 횟수(최초 1회 포함)
    - base_delay: 첫 백오프 기준
    - jitter: 랜덤 지터 범위(초)
    """

    def decorator(func: Callable):
        @functools.wraps(func)
        def wrapper(*args, **kwargs):
            last_exc: BaseException | None = None

            for attempt in range(1, retries + 1):
                try:
                    return func(*args, **kwargs)
                except retry_on as exc:
                    last_exc = exc
                    if attempt >= retries:
                        raise

                    # 지수 백오프 + 지터
                    delay = min(max_delay, base_delay * (2 ** (attempt - 1)))
                    delay = max(0.0, delay + random.uniform(-jitter, jitter))

                    if on_retry:
                        on_retry(attempt, exc, delay)

                    time.sleep(delay)

            # 이 지점에 오면 로직상 이상하지만, 타입 체커를 위한 안전장치
            assert last_exc is not None
            raise last_exc

        return wrapper

    return decorator

import time
import functools
import hashlib
import json
from dataclasses import is_dataclass, asdict
from typing import Any, Callable


def _stable_json(obj: Any) -> str:
    if is_dataclass(obj):
        obj = asdict(obj)
    return json.dumps(obj, sort_keys=True, default=str, separators=(",", ":"))


def default_cache_key(func_name: str, args: tuple, kwargs: dict) -> str:
    payload = {
        "func": func_name,
        "args": args,
        "kwargs": kwargs,
    }
    raw = _stable_json(payload).encode("utf-8")
    return hashlib.sha256(raw).hexdigest()


def ttl_cache(
    *,
    ttl_seconds: float = 30.0,
    key_fn: Callable[[str, tuple, dict], str] = default_cache_key,
    cache_none: bool = True,
):
    """동기 함수용 TTL 캐시 데코레이터(프로세스 메모리).

    - cache_none: 결과가 None일 때도 캐시할지 여부
    """

    store: dict[str, tuple[float, Any]] = {}

    def decorator(func: Callable):
        @functools.wraps(func)
        def wrapper(*args, **kwargs):
            key = key_fn(func.__name__, args, kwargs)
            now = time.time()

            if key in store:
                expires_at, value = store[key]
                if now < expires_at:
                    return value
                store.pop(key, None)

            value = func(*args, **kwargs)
            if value is None and not cache_none:
                return value

            store[key] = (now + ttl_seconds, value)
            return value

        return wrapper

    return decorator

import functools
from concurrent.futures import ThreadPoolExecutor, TimeoutError as FutureTimeout
from typing import Callable, TypeVar

T = TypeVar("T")


def timeout(*, seconds: float, executor: ThreadPoolExecutor | None = None):
    """동기 함수용 타임아웃 데코레이터.

    주의: 타임아웃이 나도 작업 스레드가 백그라운드에서 계속 실행될 수 있습니다.
    I/O 라이브러리의 자체 타임아웃과 병행하는 것을 권장합니다.
    """

    def decorator(func: Callable[..., T]):
        @functools.wraps(func)
        def wrapper(*args, **kwargs) -> T:
            local_executor = executor or ThreadPoolExecutor(max_workers=1)
            owns = executor is None

            try:
                fut = local_executor.submit(func, *args, **kwargs)
                return fut.result(timeout=seconds)
            except FutureTimeout as exc:
                raise TimeoutError(f"Timed out after {seconds} seconds") from exc
            finally:
                if owns:
                    local_executor.shutdown(wait=False, cancel_futures=True)

        return wrapper

    return decorator

import urllib.request


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


@ttl_cache(ttl_seconds=10.0)
@retry(retries=4, base_delay=0.2, max_delay=2.0, jitter=0.2, on_retry=log_retry)
@timeout(seconds=1.0)
def fetch_text(url: str) -> str:
    # 라이브러리 자체 타임아웃도 함께 거는 게 더 안전합니다.
    with urllib.request.urlopen(url, timeout=0.8) as resp:
        return resp.read().decode("utf-8")