목차
FastAPI 트랜잭션 관리
FastAPI는 SQLAlchemy와 같은 ORM을 활용해 트랜잭션을 쉽게 처리할 수 있다.
SQLAlchemy는 session.begin() 또는 async_session.begin() 같은 컨텍스트 매니저로 트랜잭션을 안전하게 관리한다.
•
동기 방식
with Session(engine) as session:
with sesseion.begin():
# 트랜잭션 내 로직 수행
Python
복사
•
비동기 방식
async with AsyncSession(engine) as session:
async with Session.begin():
# 트랜잭션 내 로직 수행
Python
복사
•
commit(), rollback()을 직접 호출
with Session(engine) as session:
try:
session.begin()
session.add(...)
...
session.commit() # 수동 커밋
except:
session.rollback() # 수동 롤백
raise
Python
복사
FastAPI 적용 예시
DB 세션 / 모델 준비
•
model
from sqlalchemy import Column, Integer, String
from sqlalchemy.orm import declarative_base
Base = declarative_base()
class User(Base):
__tablename__ = "users"
id = Column(Integer, primary_key=True)
name = Column(String, nullable=False)
Python
복사
•
session
from sqlalchemy import create_engine
from sqlalchemy.ext.asyncio import create_async_engine, AsyncSession
from sqlalchemy.orm import sessionmaker
# 비동기용
ASYNC_DATABASE_URL = "db주소"
async_engine = create_async_engine(ASYNC_DATABASE_URL, echo=True)
AsyncSessionLocal = sessioinmaker(bind=async_engine, class_=AsyncSession, expire_on_commit=False)
# 의존성
async def get_async_db():
db = AsyncSessonLocal()
try:
yield db
finally:
await db.close()
Python
복사
로직에 적용
•
router
◦
라우터에서는 크게 session만 DI로 받아옴
@router.post("/async/users")
async def create_async_user(name: str, db: AsyncSession = Depends(get_async_db)):
try:
return await create_user_service(db, name)
except Exception as e:
raise HTTPException(status_code=400, detail=str(e))
Python
복사
•
service
◦
서비스 전체에서 트랜잭션 범위로 잡음
from sqlalchemy.ext.asyncio import AsyncSession
from repositories.user_repository import create_user
async def create_user_service(session: AsyncSession, name: str):
async with session.begin():
if name == "error":
raise ValueError("Invalid name")
await create_user(session, name)
# commit은 자동
Python
복사
•
repository
from sqlalchemy.ext.asyncio import AsyncSession
from models import User
async def create_user(session: AsyncSession, name: str):
user = User(name=name)
session.add(user) # 비동기지만 add는 await 필요 없음
Python
복사
데코레이션 적용 예시
with 문을 사용하면 인덴트가 뒤로 밀리기 때문에 코드가 좀 지저분해질 수 있다.
동기 방식
•
데코레이터 정의
# decorators/transactional.py
def transactional():
def decorator(func):
def wrapper(session, *args, **kwargs):
with session.begin():
return func(session, *args, **kwargs)
return wrapper
return decorator
Python
복사
•
서비스에서 데코레이터 사용
from decorators.transactional import transactional
@transactional()
def create_user_service(session: Session, name: str):
if name == "error":
raise ValueError("Invalid name")
user_repo.create_user(session, name)
...
Python
복사
비동기 방식
•
데코레이터 정의
import functools
def async_transactional():
def decorator(func):
@functools.wraps(func)
async def wrapper(session, *args, **kwargs):
async with session.begin():
return await func(session, *args, **kwargs):
Python
복사
•
서비스에서 데코레이터 사용
@async_transactional()
async def create_user_service(session: AsyncSession, name: str):
# 인덴트 없이 로직만 깔끔하게
if name == "error":
raise ValueError("Invalid name")
await user_repo.create_user(session, name)
...
Python
복사
