데코레이터+컨텍스트 매니저로 리소스 누수 0

from contextlib import ExitStack

class FakeConn:
    def __init__(self, name: str):
        self.name = name
        self.closed = False

    def close(self):
        self.closed = True


def open_conn(name: str) -> FakeConn:
    return FakeConn(name)


def do_work():
    with ExitStack() as stack:
        c1 = stack.enter_context(_closing(open_conn("primary")))
        c2 = stack.enter_context(_closing(open_conn("replica")))
        # 여기서 예외가 나도 c2, c1 순서로 close 보장
        return (c1.name, c2.name)


from contextlib import contextmanager

@contextmanager
def _closing(obj):
    try:
        yield obj
    finally:
        obj.close()

from __future__ import annotations

from contextlib import contextmanager
from functools import wraps
from typing import Callable, Iterator, TypeVar, ParamSpec

P = ParamSpec("P")
R = TypeVar("R")


def with_resource(cm_factory: Callable[P, "_CM" ]):
    def decorator(fn: Callable[P, R]) -> Callable[P, R]:
        @wraps(fn)
        def wrapper(*args: P.args, **kwargs: P.kwargs) -> R:
            with cm_factory(*args, **kwargs):
                return fn(*args, **kwargs)
        return wrapper
    return decorator


class _CM:
    def __enter__(self): ...
    def __exit__(self, exc_type, exc, tb): ...


@contextmanager
def request_scope(*args, **kwargs) -> Iterator[None]:
    # 예: request_id 세팅, 임시 디렉터리 생성, 로깅 컨텍스트 주입 등
    try:
        yield
    finally:
        # 반드시 정리
        pass


@with_resource(request_scope)
def handler(user_id: str) -> str:
    return f"ok:{user_id}"

from __future__ import annotations

from contextlib import contextmanager
from functools import wraps
from typing import Callable, Iterator, TypeVar, ParamSpec

P = ParamSpec("P")
R = TypeVar("R")


def inject_resource(
    resource_cm: Callable[P, Iterator[object]],
    *,
    param: str,
):
    def decorator(fn: Callable[P, R]) -> Callable[P, R]:
        @wraps(fn)
        def wrapper(*args: P.args, **kwargs: P.kwargs) -> R:
            if param in kwargs:
                raise TypeError(f"'{param}' is managed by decorator")

            with resource_cm(*args, **kwargs) as resource:
                kwargs[param] = resource
                return fn(*args, **kwargs)

        return wrapper

    return decorator


class Conn:
    def __init__(self):
        self.closed = False

    def close(self):
        self.closed = True


@contextmanager
def conn_cm(*args, **kwargs) -> Iterator[Conn]:
    c = Conn()
    try:
        yield c
    finally:
        c.close()


@inject_resource(conn_cm, param="conn")
def get_user(user_id: str, *, conn: Conn) -> dict:
    # conn 사용
    return {"user_id": user_id, "closed": conn.closed}

from contextlib import contextmanager
from typing import Iterator

class TxConn:
    def __init__(self):
        self.closed = False
        self.in_tx = False

    def begin(self):
        self.in_tx = True

    def commit(self):
        self.in_tx = False

    def rollback(self):
        self.in_tx = False

    def close(self):
        self.closed = True


@contextmanager
def transactional_conn() -> Iterator[TxConn]:
    c = TxConn()
    c.begin()
    try:
        yield c
        c.commit()
    except Exception:
        c.rollback()
        raise
    finally:
        c.close()

import threading
from contextlib import contextmanager
from typing import Iterator

@contextmanager
def acquired(lock: threading.Lock, *, timeout: float | None = None) -> Iterator[None]:
    ok = lock.acquire(timeout=timeout) if timeout is not None else lock.acquire()
    if not ok:
        raise TimeoutError("lock acquire timeout")
    try:
        yield
    finally:
        lock.release()

import time
from typing import Callable, TypeVar

R = TypeVar("R")


def retry(times: int, *, delay: float) -> Callable[[Callable[[], R]], Callable[[], R]]:
    def decorator(fn: Callable[[], R]) -> Callable[[], R]:
        def wrapper() -> R:
            last_exc: Exception | None = None
            for _ in range(times):
                try:
                    return fn()
                except Exception as e:
                    last_exc = e
                    time.sleep(delay)
            assert last_exc is not None
            raise last_exc
        return wrapper
    return decorator


@retry(3, delay=0.2)
def do_tx_job() -> str:
    # 재시도 1회마다 transactional_conn 컨텍스트가 새로 열리고 닫힘
    with transactional_conn() as conn:
        # 작업 수행
        return f"done:{conn.closed}"  # 함수 반환 시점에는 아직 close 전

import logging
from contextlib import contextmanager
from typing import Iterator

logger = logging.getLogger(__name__)

@contextmanager
def logged_resource(name: str) -> Iterator[None]:
    logger.info("resource acquire: %s", name)
    try:
        yield
    finally:
        logger.info("resource release: %s", name)

from __future__ import annotations

import tempfile
import threading
from contextlib import ExitStack, contextmanager
from dataclasses import dataclass
from typing import Iterator

@dataclass
class Scope:
    tmpdir: str
    lock: threading.Lock
    conn: TxConn


@contextmanager
def scope_cm() -> Iterator[Scope]:
    lock = threading.Lock()

    with ExitStack() as stack:
        # tmpdir
        tmp = stack.enter_context(tempfile.TemporaryDirectory())

        # lock acquired
        stack.enter_context(acquired(lock, timeout=1.0))

        # transactional conn
        conn = stack.enter_context(transactional_conn())

        yield Scope(tmpdir=tmp.name, lock=lock, conn=conn)


def inject_scope(param: str = "scope"):
    def decorator(fn):
        from functools import wraps

        @wraps(fn)
        def wrapper(*args, **kwargs):
            if param in kwargs:
                raise TypeError(f"'{param}' is managed by decorator")
            with scope_cm() as scope:
                kwargs[param] = scope
                return fn(*args, **kwargs)
        return wrapper
    return decorator


@inject_scope("scope")
def run_job(job_id: str, *, scope: Scope) -> str:
    # scope.tmpdir, scope.conn 등을 안전하게 사용
    return f"job:{job_id}:tmp={scope.tmpdir}"