Python 데코레이터+컨텍스트 매니저로 자동 롤백

from contextlib import contextmanager
from typing import Iterator, Optional


@contextmanager
def transaction(conn, *, read_only: bool = False) -> Iterator[object]:
    """커넥션 단위 트랜잭션 컨텍스트.

    - 성공: commit
    - 예외: rollback
    - 항상: 커서/리소스 정리(필요 시)

    read_only는 예시이며, 실제로는 DB별로 설정 방식이 다릅니다.
    """

    if read_only:
        # DB마다 read only 설정 방식이 다르므로 예시로만 둡니다.
        # 예: PostgreSQL은 BEGIN READ ONLY 같은 형태
        pass

    try:
        yield conn
        if not read_only:
            conn.commit()
        else:
            # 읽기 전용이면 커밋을 생략하거나, DB 정책에 맞게 처리
            conn.rollback()  # 일부 드라이버는 read-only라도 트랜잭션 종료가 필요
    except Exception:
        conn.rollback()
        raise

import functools
from typing import Callable, TypeVar, Any

R = TypeVar("R")


def transactional(get_conn: Callable[[], Any], *, read_only: bool = False):
    """함수 단위 트랜잭션 데코레이터.

    get_conn: 커넥션을 꺼내는 함수(예: pool.getconn, DI 컨테이너)
    """

    def decorator(fn: Callable[..., R]) -> Callable[..., R]:
        @functools.wraps(fn)
        def wrapper(*args, **kwargs) -> R:
            conn = get_conn()
            try:
                with transaction(conn, read_only=read_only):
                    return fn(*args, **kwargs)
            finally:
                # 풀을 쓴다면 반환 로직이 들어갑니다.
                # 예: pool.putconn(conn)
                pass

        return wrapper

    return decorator

# 예: 전역 풀 또는 DI로 주입된 풀

def get_conn():
    # pool.getconn() 같은 로직
    return conn


@transactional(get_conn)
def transfer_money(from_user_id: int, to_user_id: int, amount: int) -> None:
    # 여기서는 conn을 직접 받지 않는 구조도 가능하지만,
    # 예제에서는 단순화를 위해 내부에서 전역/스레드로컬을 쓴다고 가정합니다.
    # 실제로는 conn을 인자로 받거나, repository가 conn을 받도록 구성하는 편이 안전합니다.

    # debit
    # credit
    # ledger insert
    # 중간에 예외가 나면 자동 rollback
    pass

from contextvars import ContextVar
from contextlib import contextmanager

_in_tx: ContextVar[bool] = ContextVar("_in_tx", default=False)


@contextmanager
def transaction_scoped(conn):
    token = None
    if _in_tx.get():
        # 이미 트랜잭션 안이면 참여만 하고 커밋/롤백 책임은 상위에 둡니다.
        yield conn
        return

    try:
        token = _in_tx.set(True)
        yield conn
        conn.commit()
    except Exception:
        conn.rollback()
        raise
    finally:
        if token is not None:
            _in_tx.reset(token)

import functools


def transactional_scoped(get_conn):
    def decorator(fn):
        @functools.wraps(fn)
        def wrapper(*args, **kwargs):
            conn = get_conn()
            try:
                with transaction_scoped(conn):
                    return fn(*args, **kwargs)
            finally:
                pass

        return wrapper

    return decorator

from sqlalchemy.orm import sessionmaker

SessionLocal = sessionmaker(bind=engine, autoflush=False, autocommit=False)


def with_session(fn):
    import functools

    @functools.wraps(fn)
    def wrapper(*args, **kwargs):
        session = SessionLocal()
        try:
            with session.begin():
                return fn(session, *args, **kwargs)
        except Exception:
            # begin 컨텍스트가 rollback을 수행하지만,
            # 여기서 로깅/추적을 추가할 수 있습니다.
            raise
        finally:
            session.close()

    return wrapper


@with_session
def create_user(session, email: str) -> int:
    user = User(email=email)
    session.add(user)
    session.flush()  # PK 필요 시
    return user.id