tenacity를 이용한 retry 사용 방법 정리

1. 개요

tenacity는 Python에서 함수 또는 코드 블록을 자동으로 재시도(retry)할 수 있도록 도와주는 라이브러리입니다. 네

트워크 요청, 데이터베이스 연결 등 실패할 가능성이 있는 작업을 안정적으로 수행할 수 있도록 합니다.

 

2. 설치

pip install tenacity

 

3. 기본 사용법

   1) 간단한 재시도 적용

      - 예외가 발생할 경우 기본적으로 무한 재시도를 수행

from tenacity import retry

@retry
def unstable_function():
    print("실행 중...")
    raise Exception("에러 발생!")

unstable_function()

 

 

4. 고급 사용법

   1) 최대 재시도 횟수 설정

from tenacity import retry, stop_after_attempt

@retry(stop=stop_after_attempt(3))
def unstable_function():
    print("실행 중...")
    raise Exception("에러 발생!")

unstable_function()

 

   2) 재시도 간격 설정 (대기 시간 설정)

from tenacity import retry, wait_fixed

@retry(wait=wait_fixed(2))  # 2초 간격으로 재시도

 

   3) 지수 백오프 (Exponential Backoff) 적용

      - 1초부터 시작하여 점점 증가하는 대기 시간을 적용한다.

from tenacity import retry, wait_exponential

@retry(wait=wait_exponential(multiplier=1, min=1, max=10))
def unstable_function():
    print("실행 중...")
    raise Exception("에러 발생!")

unstable_function()

 

   4) 특정 예외에 대해서만 재시도

      - ValueError가 발생할 때만 재시도를 수행한다.

from tenacity import retry, retry_if_exception_type

@retry(retry=retry_if_exception_type(ValueError))
def unstable_function():
    raise ValueError("ValueError 발생")

 

   5) 여러 예외 처리

      - ValueError 또는 KeyError 발생 시 재시도를 수행한다.

from tenacity import retry, retry_if_exception_type

@retry(retry=retry_if_exception_type((ValueError, KeyError)))
def unstable_function():
    raise KeyError("KeyError 발생")

 

   6) 특정 조건에 따른 재시도

      - 반환값이 None일 경우 재시도를 수행한다.

from tenacity import retry, retry_if_result

def is_none(value):
    return value is None

@retry(retry=retry_if_result(is_none))
def unstable_function():
    return None  # None이 반환되면 재시도

 

   7) 재시도 후 실행할 콜백 함수 지정

      - 재시도 전에 특정 콜백 함수를 실행하여 로그를 남긴다.

from tenacity import retry, before_sleep

def log_before_retry(retry_state):
    print(f"재시도 중: {retry_state.attempt_number}번째 시도")

@retry(before_sleep=log_before_retry)
def unstable_function():
    raise Exception("에러 발생!")

unstable_function()

 

   8) 성공 시 동작 설정

      - 재시도 후 성공 시 특정 동작을 수행한다.

from tenacity import retry, after

def log_success(retry_state):
    print(f"성공: {retry_state.attempt_number}번째 시도에서 성공")

@retry(after=log_success)
def unstable_function():
    print("성공적으로 실행됨!")
    return True

unstable_function()

 

   9) 로그 출력 설정

import logging
from tenacity import retry

logging.basicConfig(level=logging.INFO)

@retry()
def unstable_function():
    logging.info("실행 중...")
    raise Exception("에러 발생!")

unstable_function()

 

5. fastapi에서 sqlalchemy를 사용하는 경우 tenacity를 사용하는 방법

   - version을 이용한 낙관적 잠금(Optimistic Lock)을 사용

   - 엔드포인트에서 HTTPException 발생 조건의 3회 재시도

   - db 핸들링 함수에서 StaleDataError 발생 조건에서 0.3초부터 시작하여 2배씩 증가 (최대 10초), 최대 10번 재시도

from fastapi import FastAPI, HTTPException, Depends
from sqlalchemy.ext.declarative import declarative_base
from sqlalchemy import Column, Integer, String, create_engine
from sqlalchemy.orm import sessionmaker, Session
from sqlalchemy.exc import StaleDataError
from tenacity import retry, stop_after_attempt, wait_exponential, retry_if_exception_type

# 데이터베이스 설정
DATABASE_URL = "sqlite:///./test.db"
engine = create_engine(DATABASE_URL, connect_args={"check_same_thread": False})
SessionLocal = sessionmaker(autocommit=False, autoflush=False, bind=engine)
Base = declarative_base()

def custom_version_generator(current_version):
    return current_version + 1  # 현재 버전에서 1 증가

# 테이블 정의
class Item(Base):
    __tablename__ = "items"
    id = Column(Integer, primary_key=True, index=True)
    name = Column(String, index=True)
    like = Column(Integer, nullable=False, default=0)  # 좋아요 필드 추가
    version = Column(Integer, nullable=False, default=0)
    __mapper_args__ = {
        "version_id_col": version,
        "version_id_generator": custom_version_generator,
    }

Base.metadata.create_all(bind=engine)

# FastAPI 앱 생성
app = FastAPI()

def get_db():
    db = SessionLocal()
    try:
        yield db
    finally:
        db.close()

# 충돌 발생 시 exponential backoff를 적용한 재시도 로직
@retry(
    retry=retry_if_exception_type(StaleDataError),  # StaleDataError 발생 시 재시도
    wait=wait_exponential(multiplier=0.3, min=0.3, max=10),  # 0.3초부터 시작하여 2배씩 증가 (최대 10초)
    stop=stop_after_attempt(10)  # 최대 10번 재시도
)
def increment_like_with_lock(db: Session, item_id: int):
    """
    주어진 item_id에 해당하는 아이템을 찾고 like 값을 +1 증가한 후 저장
    충돌이 발생하면 롤백하고 다시 시도 (최대 10회)
    """
    item = db.query(Item).filter(Item.id == item_id).first()
    if not item:
        raise HTTPException(status_code=404, detail="Item not found")  # 아이템이 존재하지 않으면 404 반환
    item.like += 1  # like 값 증가
    try:
        db.commit()  # 변경 사항 저장
        db.refresh(item)  # 최신 데이터 반영
        return item
    except StaleDataError:
        db.rollback()  # 충돌 발생 시 롤백
        raise
    except Exception:
        db.rollback()  # 기타 예외 발생 시 롤백 후 에러 반환
        raise HTTPException(status_code=500, detail="Internal Server Error")

@app.put("/items/{item_id}/like")
@retry(
    retry=retry_if_exception_type(HTTPException),  # HTTPException 발생 시 재시도
    stop=stop_after_attempt(3),  # 최대 3회 재시도
)
def increment_like(item_id: int, db: Session = Depends(get_db)):
    """
    FastAPI 엔드포인트: 아이템의 like 값을 증가하는 API
    HTTPException이 발생하면 최대 3회 재시도
    기타 예외가 발생하면 재시도 없이 즉시 중단
    """
    return increment_like_with_lock(db, item_id)