Python 데코레이터·컨텍스트 매니저로 로깅·트랜잭션 중복 제거

서비스 코드가 커질수록 비즈니스 로직 자체보다 주변부 코드(로깅, 트랜잭션, 예외 변환, 메트릭, 재시도) 가 더 많이 보이기 시작합니다. 특히 Python에서는 함수형 스타일과 OOP가 섞여 있어, 팀마다 제각각의 방식으로 try/except/finally 블록이 늘어나기 쉽습니다.

이 글에서는 실무에서 자주 겪는 두 가지 중복을 대상으로 합니다.

요청/작업 단위의 일관된 로깅(시작/성공/실패/소요시간/상관관계 ID)
DB 작업의 트랜잭션 처리(commit/rollback/close)

그리고 이를 데코레이터와 컨텍스트 매니저로 분리해, 비즈니스 코드에서 잡음을 제거하는 방법을 코드로 보여줍니다.

관련해서 “재시도/백오프” 계층까지 포함해 설계를 확장하고 싶다면, 네트워크 호출의 표준 패턴을 다룬 글도 함께 참고하면 좋습니다: OpenAI API 429 재시도·백오프 패턴 실전 가이드

왜 데코레이터와 컨텍스트 매니저인가

데코레이터: 함수 호출 전후에 공통 관심사를 주입하기 좋습니다. 예를 들어 “요청 시작 로그”, “실패 시 예외 로깅”, “소요시간 측정” 등은 함수 경계에서 처리하는 게 자연스럽습니다.
컨텍스트 매니저: 리소스의 생명주기(획득/해제)를 보장합니다. 트랜잭션처럼 commit 또는 rollback이 반드시 실행되어야 하는 경우 with가 가장 읽기 쉽고 안전합니다.

핵심은 역할 분리입니다.

데코레이터는 관측 가능성(Observability)
컨텍스트 매니저는 자원/일관성(Consistency)

이 둘을 섞어버리면 재사용성이 떨어지고, 테스트도 어려워집니다.

문제: 중복 로깅·트랜잭션 코드가 비즈니스 로직을 덮는다

아래는 흔히 보이는 형태입니다. 함수마다 로깅 포맷이 조금씩 다르고, 트랜잭션 처리도 예외 케이스에서 누락되기 쉽습니다.

import logging
import time

logger = logging.getLogger(__name__)


def create_order(conn, user_id: int, items: list[dict]) -> int:
    start = time.perf_counter()
    logger.info("create_order start user_id=%s items=%d", user_id, len(items))

    try:
        cur = conn.cursor()
        cur.execute("BEGIN")

        # 비즈니스 로직
        cur.execute("INSERT INTO orders(user_id) VALUES (%s) RETURNING id", (user_id,))
        order_id = cur.fetchone()[0]
        for it in items:
            cur.execute(
                "INSERT INTO order_items(order_id, sku, qty) VALUES (%s, %s, %s)",
                (order_id, it["sku"], it["qty"]),
            )

        conn.commit()
        elapsed_ms = (time.perf_counter() - start) * 1000
        logger.info("create_order success order_id=%s elapsed_ms=%.2f", order_id, elapsed_ms)
        return order_id

    except Exception:
        conn.rollback()
        elapsed_ms = (time.perf_counter() - start) * 1000
        logger.exception("create_order failed elapsed_ms=%.2f", elapsed_ms)
        raise
    finally:
        # 커서/커넥션 close 등도 여기저기 섞이기 시작
        pass

이 코드는 “동작”은 하지만, 팀 규모가 커질수록 다음 문제가 누적됩니다.

로그 키가 통일되지 않아 검색/대시보드가 어려움
트랜잭션 시작/커밋/롤백의 관례가 코드마다 다름
예외 로깅이 중복되어 Sentry 같은 시스템에서 노이즈 증가
테스트에서 conn.commit() 호출 여부를 검증하기가 까다로움

1) 로깅 데코레이터: 함수 경계에서 표준화

먼저 “함수 시작/성공/실패/소요시간”을 표준화하는 데코레이터를 만듭니다.

설계 포인트:

functools.wraps로 원 함수 메타데이터 보존
로그 메시지는 문자열 포맷보다 구조화된 키를 유지(최소한 key=value)
민감정보/대용량 인자 로깅을 피하기 위한 arg_summarizer 훅 제공

from __future__ import annotations

import functools
import logging
import time
from collections.abc import Callable
from typing import Any

logger = logging.getLogger(__name__)


def default_arg_summarizer(args: tuple[Any, ...], kwargs: dict[str, Any]) -> dict[str, Any]:
    # 실무에서는 PII/토큰/비밀번호 등을 마스킹해야 합니다.
    # 여기서는 길이/타입 중심으로 안전하게 요약합니다.
    return {
        "args_len": len(args),
        "kwargs_keys": sorted(list(kwargs.keys())),
    }


def logged(
    *,
    action: str | None = None,
    arg_summarizer: Callable[[tuple[Any, ...], dict[str, Any]], dict[str, Any]] = default_arg_summarizer,
) -> Callable[[Callable[..., Any]], Callable[..., Any]]:
    def deco(fn: Callable[..., Any]) -> Callable[..., Any]:
        act = action or fn.__name__

        @functools.wraps(fn)
        def wrapper(*args: Any, **kwargs: Any) -> Any:
            start = time.perf_counter()
            ctx = arg_summarizer(args, kwargs)
            logger.info("action=%s stage=start %s", act, " ".join(f"{k}={v}" for k, v in ctx.items()))

            try:
                result = fn(*args, **kwargs)
                elapsed_ms = (time.perf_counter() - start) * 1000
                logger.info("action=%s stage=success elapsed_ms=%.2f", act, elapsed_ms)
                return result
            except Exception as e:
                elapsed_ms = (time.perf_counter() - start) * 1000
                logger.exception(
                    "action=%s stage=error elapsed_ms=%.2f exc_type=%s",
                    act,
                    elapsed_ms,
                    type(e).__name__,
                )
                raise

        return wrapper

    return deco

이제 비즈니스 함수는 로깅을 신경 쓰지 않아도 됩니다.

@logged(action="create_order")
def create_order(conn, user_id: int, items: list[dict]) -> int:
    cur = conn.cursor()
    cur.execute("INSERT INTO orders(user_id) VALUES (%s) RETURNING id", (user_id,))
    order_id = cur.fetchone()[0]
    for it in items:
        cur.execute(
            "INSERT INTO order_items(order_id, sku, qty) VALUES (%s, %s, %s)",
            (order_id, it["sku"], it["qty"]),
        )
    return order_id

하지만 아직 트랜잭션은 남아 있습니다. 로깅은 “함수 경계”, 트랜잭션은 “리소스 경계”에서 처리하는 게 더 자연스럽습니다.

2) 트랜잭션 컨텍스트 매니저: commit/rollback을 강제

트랜잭션은 아래 조건을 만족해야 합니다.

정상 종료 시 commit
예외 발생 시 rollback
커서/세션/커넥션 해제 정책을 한 곳에서 통제

DB 드라이버마다 API가 다르므로, 여기서는 가장 흔한 패턴(커넥션이 commit, rollback을 가짐)을 기준으로 작성합니다.

from __future__ import annotations

from contextlib import contextmanager
from typing import Any, Iterator


@contextmanager
def transaction(conn: Any) -> Iterator[Any]:
    """간단한 트랜잭션 컨텍스트.

    - with 블록이 정상 종료되면 commit
    - 예외가 발생하면 rollback 후 예외 재발생
    """
    try:
        yield conn
        conn.commit()
    except Exception:
        conn.rollback()
        raise

적용 예시는 다음과 같습니다.

@logged(action="create_order")
def create_order(conn, user_id: int, items: list[dict]) -> int:
    with transaction(conn):
        cur = conn.cursor()
        cur.execute("INSERT INTO orders(user_id) VALUES (%s) RETURNING id", (user_id,))
        order_id = cur.fetchone()[0]
        for it in items:
            cur.execute(
                "INSERT INTO order_items(order_id, sku, qty) VALUES (%s, %s, %s)",
                (order_id, it["sku"], it["qty"]),
            )
        return order_id

이 조합의 장점은 명확합니다.

함수는 “주문 생성”만 말한다
트랜잭션 정책은 transaction()이 전담한다
로깅 정책은 @logged가 전담한다

3) 실무형 개선: 중첩 트랜잭션, savepoint, read-only

현업에서는 다음 요구가 금방 등장합니다.

서비스 함수가 다른 서비스 함수를 호출하면서 트랜잭션이 중첩됨
일부 구간만 롤백하고 싶어 savepoint가 필요함
조회 API는 read-only로 분리하고 싶음

드라이버/ORM에 따라 지원 방식이 다르지만, “중첩 호출에서 바깥 트랜잭션만 커밋” 같은 정책은 공통적으로 중요합니다.

간단한 방식은 “이미 트랜잭션이 열려 있으면 commit/rollback을 하지 않는다”라는 플래그를 두는 것입니다. 아래는 개념 예시입니다.

from contextlib import contextmanager
from typing import Iterator, Any


@contextmanager
def transaction(conn: Any) -> Iterator[Any]:
    # 예: psycopg2는 autocommit 플래그로 트랜잭션 동작이 달라집니다.
    # 여기서는 conn에 "_tx_depth"가 있다고 가정해 개념만 보여줍니다.
    depth = getattr(conn, "_tx_depth", 0)
    setattr(conn, "_tx_depth", depth + 1)

    try:
        yield conn
        # 최상위 트랜잭션만 commit
        if depth == 0:
            conn.commit()
    except Exception:
        # 최상위 트랜잭션만 rollback
        if depth == 0:
            conn.rollback()
        raise
    finally:
        setattr(conn, "_tx_depth", depth)

실제로는 커넥션 객체에 임의 속성을 붙이기보다, 세션 래퍼(예: DbSession)를 만들어 깊이/락/메트릭을 관리하는 편이 안전합니다.

4) 데코레이터와 컨텍스트 매니저의 책임 경계

둘을 같이 쓰다 보면 “트랜잭션도 데코레이터로 감싸면 더 짧지 않나?”라는 유혹이 생깁니다. 가능은 하지만, 실무에서는 아래 이유로 컨텍스트 매니저가 더 낫습니다.

트랜잭션 범위를 함수 전체로 강제하면, 일부 구간만 트랜잭션을 열고 닫는 최적화가 어려움
함수 내부에서 여러 DB를 다루거나, 외부 API 호출을 트랜잭션 밖으로 빼고 싶은 경우가 많음
with는 “여기부터 여기까지”가 코드로 드러나 리뷰가 쉬움

반대로 로깅은 함수 경계에서 “일괄” 적용하는 편이 일관성이 높고, 누락도 줄어듭니다.

5) 예외 처리 전략: 로깅 중복과 예외 변환

logger.exception()은 스택트레이스를 남기기 때문에 강력하지만, 계층별로 중복 로깅하면 관측 시스템에서 같은 오류가 여러 번 찍힙니다.

권장 패턴:

경계 레이어(핸들러/작업 런너) 에서만 스택트레이스 로깅
도메인 레이어에서는 “의미 있는 예외로 변환”만 하고 로깅은 최소화

만약 지금 글의 @logged를 모든 도메인 함수에 붙인다면, 예외 로깅이 과해질 수 있습니다. 실무에서는 아래처럼 옵션을 둡니다.

def logged(*, action: str | None = None, log_exceptions: bool = True):
    def deco(fn):
        ...
        def wrapper(*args, **kwargs):
            ...
            except Exception as e:
                if log_exceptions:
                    logger.exception(...)
                else:
                    logger.warning("action=%s stage=error exc_type=%s", act, type(e).__name__)
                raise
        return wrapper
    return deco

“어디서 스택트레이스를 남길지”를 정하면, 알람 품질이 크게 좋아집니다.

6) 비동기 환경: `async` 함수와 `async with`

FastAPI, asyncio 작업자, 비동기 DB 드라이버를 쓰면 async def와 async with가 필요합니다. 패턴은 동일하지만 구현이 달라집니다.

로깅: async 함수도 감싸는 데코레이터 필요
트랜잭션: @asynccontextmanager 사용

import functools
import time
import logging
from contextlib import asynccontextmanager
from typing import Any, Callable, Awaitable

logger = logging.getLogger(__name__)


def async_logged(action: str | None = None) -> Callable[[Callable[..., Awaitable[Any]]], Callable[..., Awaitable[Any]]]:
    def deco(fn: Callable[..., Awaitable[Any]]) -> Callable[..., Awaitable[Any]]:
        act = action or fn.__name__

        @functools.wraps(fn)
        async def wrapper(*args: Any, **kwargs: Any) -> Any:
            start = time.perf_counter()
            logger.info("action=%s stage=start", act)
            try:
                result = await fn(*args, **kwargs)
                elapsed_ms = (time.perf_counter() - start) * 1000
                logger.info("action=%s stage=success elapsed_ms=%.2f", act, elapsed_ms)
                return result
            except Exception as e:
                elapsed_ms = (time.perf_counter() - start) * 1000
                logger.exception("action=%s stage=error elapsed_ms=%.2f exc_type=%s", act, elapsed_ms, type(e).__name__)
                raise

        return wrapper

    return deco


@asynccontextmanager
async def async_transaction(session: Any):
    try:
        yield session
        await session.commit()
    except Exception:
        await session.rollback()
        raise

비동기에서는 특히 “트랜잭션 안에서 외부 API 호출을 기다리는” 실수를 조심해야 합니다. 락이 오래 잡히고, 커넥션 풀이 고갈되며, 전체 지연이 전파됩니다. 외부 호출 재시도 설계는 별도 주제로 깊게 다루는 게 좋습니다. 예: Python httpx ReadTimeout·ConnectError 재시도 설계

7) 테스트 전략: 커밋/롤백이 호출되는지 검증

패턴을 도입하면 테스트가 쉬워집니다. 비즈니스 함수는 DB API 호출만 신경 쓰고, 트랜잭션은 컨텍스트 매니저가 보장합니다.

간단히 unittest.mock으로 검증할 수 있습니다.

from unittest.mock import MagicMock
import pytest


def test_transaction_commit_on_success():
    conn = MagicMock()
    with transaction(conn):
        pass
    conn.commit.assert_called_once()
    conn.rollback.assert_not_called()


def test_transaction_rollback_on_error():
    conn = MagicMock()
    with pytest.raises(ValueError):
        with transaction(conn):
            raise ValueError("boom")
    conn.rollback.assert_called_once()

로깅은 “로그가 남는지” 자체보다, 필수 키(action, stage, elapsed) 가 누락되지 않는지 정도만 얇게 테스트하는 편이 유지보수에 유리합니다.

8) 운영에서 체감되는 효과

이 구조를 적용하면 코드 라인이 줄어드는 것보다, 운영에서 다음이 크게 좋아집니다.

장애 시 로그가 action 기준으로 묶여 탐색이 쉬움
트랜잭션 누락/중복 커밋 같은 실수가 줄어듦
신규 팀원이 비즈니스 로직을 읽을 때 잡음이 줄어 온보딩이 빨라짐

또한 로깅/트랜잭션을 표준화해두면, 이후에 메트릭(예: Prometheus), 트레이싱(예: OpenTelemetry), 재시도/백오프 정책을 경계 레이어에만 추가해도 전체에 일관되게 적용할 수 있습니다.

마무리: “관심사 분리”를 코드 규칙으로 고정하자

정리하면 다음 규칙이 실무에서 가장 안전합니다.

로깅/측정은 데코레이터로 함수 경계에 적용
트랜잭션/리소스 관리는 컨텍스트 매니저로 범위를 명시
스택트레이스 로깅은 경계 레이어로 모아 중복을 줄임

이 패턴을 한 번 팀 규칙으로 정해두면, 이후 기능이 늘어도 코드베이스의 “형태”가 무너지지 않습니다. 결과적으로 장애 대응 속도와 변경 안전성이 함께 올라갑니다.