데이터베이스를 사용하는 애플리케이션을 작성할 때 가장 까다로운 지점 중 하나는 트랜잭션의 범위를 어디서부터 어디까지로 설정할 것인가입니다. 단순히 데이터를 읽고 쓰는 기능을 넘어, 비즈니스 로직의 원자성을 보장하면서도 코드의 결합도를 낮추는 것은 결코 쉬운 일이 아닙니다. 흔히 시도되는 몇 가지 방법들과 그 한계, 그리고 이를 해결하기 위한 Unit of Work(UoW) 패턴에 대해 살펴봅니다.
가장 먼저 떠오르는 직관적인 방법은 리포지토리의 각 메서드에 트랜잭션을 입히는 것입니다.
class SongRepository:
@transaction
def confirm(self, song_id):
# 세션 관리와 트랜잭션 시작/종료가 데코레이터 내부에서 처리됨
song = self.session.get(Song, song_id)
song.status = "CONFIRMED"
return song
# 호출부
song_repo.confirm(song_id)
team_repo.create_team(song_id) # 별도의 트랜잭션으로 실행됨
Python
복사
이 방식은 사용 범위가 명확하고 반복되는 코드가 적다는 장점이 있습니다. 하지만 비즈니스 로직이 복잡해질수록 치명적인 단점이 드러납니다. 우선, 하나의 유즈케이스에서 여러 번의 리포지토리 호출이 일어날 때마다 트랜잭션을 새로 열고 닫으면서 속도 병목이 발생합니다(ref1).
도메인 로직의 파편화도 문제입니다. 여러 작업을 하나의 트랜잭션으로 묶으려다 보면 결국 리포지토리 메서드 안에 저장 관련 논리가 침투하게 되고, 결과적으로 도메인 로직이 엔티티와 리포지토리에 분산되어 인터페이스로서의 의미를 상실하게 됩니다.
또한, 논리적인 작업 단위에 대한 롤백이 불가능해집니다(ref2). 예를 들어 '곡이 확정되면 팀을 만든다'는 로직이 있을 때, 팀을 만드는 단계에서 실패하더라도 이미 확정된 곡의 정보는 데이터베이스에 영속화되어 버리는 데이터 부정합이 발생합니다. 이런 상황에서는 곡 확정까지 실패하는 것이 사용자 입장에서나 데이터 관리 측면에서나 더 낫습니다.
이를 피하기 위해 필요한 순간에만 수동으로 연결을 얻는 방식을 사용하기도 합니다.
def confirm_and_create_team_usecase(song_id):
# 1. 곡 확정
with Session() as session:
song = session.get(Song, song_id)
song.status = "CONFIRMED"
session.commit()
# 2. 외부 서비스 호출 (예: 팀 생성 알림 발송 - 트랜잭션을 물고 있지 않아 안전함)
response = requests.post("https://notify.com/api", json={"song_id": song_id})
# 3. 팀 생성
with Session() as session:
new_team = Team(song_id=song_id, name=f"Team_{song_id}")
session.add(new_team)
session.commit()
Python
복사
이 방법은 데이터베이스 연결 시간을 최소화할 수 있지만, 코드가 장황해지고 "곡 확정 시 팀 생성"과 같은 원자적 작업 단위를 보장할 수 없다는 태생적 한계를 그대로 가집니다(ref2).
웹 프레임워크인 FastAPI의 의존성 주입(DI)을 활용해 세션의 수명 주기를 요청(Request) 단위로 관리하는 방법도 흔히 쓰입니다.
@router.post('/confirm/{song_id}')
def confirm_song_endpoint(song_id: str, session = Depends(get_session)):
# 요청의 시작부터 끝까지 하나의 세션이 유지됨
song = session.get(Song, song_id)
song.status = "CONFIRMED"
new_team = Team(song_id=song_id)
session.add(new_team)
session.commit()
Python
복사
하지만 이 방식은 컨트롤러 계층이 데이터베이스 세션의 범위를 결정하게 만든다는 설계상의 결함이 있습니다. 비즈니스 로직의 호흡과 상관없이 요청 전체에 걸쳐 연결을 유지하므로, 운이 나쁘면 불필요하게 긴 시간 동안 커넥션을 점유하게 되어 자원 낭비를 초래합니다.
이러한 모든 문제를 우아하게 해결하는 것이 바로 Unit of Work 패턴입니다. UoW는 비즈니스 트랜잭션 동안 발생하는 모든 변경 사항을 추적하고, 작업이 완료되는 순간 이를 분석해 데이터베이스에 단 한 번의 효율적인 처리를 수행합니다(ref3).
class SqlAlchemyUnitOfWork:
def __init__(self, session_factory):
self._session_factory = session_factory
def __enter__(self):
self.session = self._session_factory()
self.songs = SongRepository(self.session)
self.teams = TeamRepository(self.session)
self.transaction = self.session.begin()
return self
def __exit__(self, exc_type, exc_val, exc_tb):
if exc_type is None:
self.session.commit()
else:
self.session.rollback()
self.session.close()
class ConfirmSongUseCase:
def __init__(self, uow: SqlAlchemyUnitOfWork):
self.uow = uow
def __call__(self, song_id: str):
# 모든 작업이 하나의 트랜잭션(UoW) 내부에서 일어남
with self.uow:
song = self.uow.songs.find_by_id(song_id)
song.confirm() # 상태 변경 (메모리상)
team = Team.create_by(song)
self.uow.teams.save(team)
# 블록을 정상적으로 나갈 때만 commit, 예외 발생 시 자동 rollba
Python
복사
UoW를 도입하면 리포지토리는 오직 '객체 컬렉션'으로서의 역할에만 집중할 수 있습니다. 트랜잭션 제어권이 유즈케이스 단위로 옮겨오면서 '곡 확정'과 '팀 생성'은 비로소 하나의 논리적 작업 단위로 묶이게 됩니다. 개발자는 데이터베이스의 기술적인 제약에서 벗어나, 비즈니스 규칙을 온전히 도메인 모델과 유즈케이스에 녹여낼 수 있게 됩니다.
parse me : 언젠가 이 메모에 쓰이면 좋을 것 같은 재료들입니다.
1.
None
from : 이 메모에 쓰인 생각을 만든 과거의 생각들과 연관관계를 설명합니다.
1.
•
앞의 글은 FastAPI의 의존성 주입(DI)의 편의성과 필요한 순간에만 수동으로 연결을 얻는 엄격한 방식의 중용으로 리포지토리의 각 메서드에 트랜잭션을 입히는 방법이 가장 나은 방법이라고 평가한다.
•
하지만 이 글은 아래 논의들(DB 연결, Aggregate의 의미)을 모두 고려했을 때 Unit of Work 패턴이 가장 나은 방식이라는 결론을 내린다.
2.
•
이 글에서 트랜잭션의 범위를 어디서부터 어디까지로 설정할 것인가 논할 때 앞의 글에서 강조하는 연결 풀의 점유 범위 및 시간을 고려한다.
3.
•
"팀원이 0명이면 팀은 해체되어야 한다"는 논리적인 명령을 내리면, UoW는 팀원 삭제와 팀 상태 변경을 하나의 트랜잭션으로 묶어 물리적인 정합성을 보장한다. 데이터베이스에 저장되는 과정에서 일부만 반영되거나 순서가 뒤섞이면 이런 불변식은 깨지고 만다.
•
변경 단위가 2~3개의 애그리게이트에 걸쳐 있는 경우에도 마찬가지다.
4.
•
중형 이상의 프로젝트는 repository 데코레이터 대신 UoW를 사용하도록 프롬프트를 수정했다.
supplementary : 이 메모에 작성된 생각을 뒷받침하는 새로운 메모입니다.
1.
None
opposite : 이 메모에 작성된 생각과 대조되는 새로운 메모입니다.
1.
None
to : 이 메모에 작성된 생각으로부터 발전된 생각의 메모입니다.
1.
None
ref : 생각에 참고한 자료입니다.
영구메모 템플릿 버전 2025.11.16