안녕하세요! 개발자 여러분, 안정적인 API를 구축하는 것은 언제나 중요한 과제입니다. 특히 데이터베이스의 기존 데이터를 업데이트할 때, 특정 필드의 중복을 방지하는 로직은 흔히 마주치는 시나리오입니다. 오늘 다룰 에러는 Python, SQLModel, FastAPI 환경에서 데이터를 업데이트하던 중 발생한 "Dit item bestaat al." (이 항목이 이미 존재합니다.)라는 메시지와 관련된 문제입니다. 이 에러의 원인을 깊이 파고들어, 깔끔한 해결 방법을 제시하고 향후 유사한 문제를 방지하기 위한 팁까지 알아보겠습니다.
1. 에러 발생 상황
사용자의 목표는 특정 item_id에 해당하는 항목의 title을 업데이트하는 API를 만드는 것이었습니다. 이때 핵심 요구사항은 새로운 title과 foreign_key_id의 조합이 데이터베이스에 이미 존재하는 다른 항목과 중복되어서는 안 된다는 것이었습니다. 만약 중복된다면 409 Conflict 에러를 발생시켜야 했습니다.
다음은 문제가 발생한 코드 스니펫입니다.
@app.patch(
"/update_item/{item_id}",
response_model=ItemPublic,
status_code=200,
)
def update_item(
*,
session: Session = Depends(get_session),
item_id: UUID,
item: UpdateItem,
):
"""This function (api) is to update item."""
db_item = session.get(Item, item_id)
if not db_item :
raise HTTPException(
status_code=404, detail="Dit item bestaat niet."
)
item_data = item.model_dump(exclude_unset=True)
db_item.sqlmodel_update(item_data)
check_if_item_exists = session.exec(
select(Item)
.where(Item.title == item.title)
.where(Item.foreign_key_id == item.foreign_key_id)
).first()
if check_if_item_exists:
raise HTTPException(
status_code=409, detail="Dit item bestaat al."
)
session.add(db_item)
session.commit()
session.refresh(db_item)
return db_item
이 코드를 실행했을 때, 특정 항목을 업데이트하려고 하면 다음과 같은 에러 메시지와 함께 요청이 실패했습니다.
{
"detail": "Dit item bestaat al."
}
사용자는 새로운 title과 foreign_key_id가 이미 다른 항목에 의해 추가되었는지 확인하고 싶었지만, 자신의 항목을 업데이트하는 경우에도 이 에러가 발생하는 것에 혼란을 겪었습니다.
2. 명확한 발생 원인
이 문제의 핵심은 중복 확인 로직이 현재 업데이트하려는 항목 자체를 중복으로 간주하고 있다는 점입니다.
- 인메모리(In-Memory) 업데이트:
db_item.sqlmodel_update(item_data)라인에서 데이터베이스에서 가져온db_item객체는 메모리상에서 업데이트됩니다. 즉, 이 시점에서db_item.title은 요청 본문에서 전달된 새로운item.title값을 가지게 됩니다. - 중복 확인 쿼리: 그 다음 실행되는
check_if_item_exists쿼리를 살펴보겠습니다.select(Item) .where(Item.title == item.title) .where(Item.foreign_key_id == item.foreign_key_id)이 쿼리는 데이터베이스에서
item.title(요청 본문의 새 제목)과item.foreign_key_id를 만족하는 항목을 찾습니다. 문제가 되는 부분은 이 쿼리가 현재 업데이트하려는db_item자신을 찾아낸다는 것입니다.db_item은 이미 메모리상에서 요청 본문의item.title과item.foreign_key_id를 가지고 있기 때문입니다. - 불필요한 에러 발생: 결과적으로, 쿼리는 항상 현재 업데이트하려는 항목(
db_item)을 찾아내고check_if_item_exists는None이 아니게 되어,409 Conflict에러가 발생하게 됩니다. 이는 사용자가 의도한 “다른 항목과의 중복 확인”이 아닌, “자기 자신과의 중복 확인”이 되어버린 것입니다.
따라서, 중복 확인 로직은 현재 업데이트하려는 항목을 제외한 ‘다른’ 항목들 중에서만 중복을 찾아야 합니다.
3. 해결 방법 및 코드 예시
이 문제를 해결하는 핵심은 중복 확인 쿼리에 현재 업데이트하려는 항목의 id를 제외하는 조건을 추가하는 것입니다. item_id는 Path Parameter로 전달받는 UUID이므로, 이를 활용하여 쿼리를 수정할 수 있습니다.
수정된 check_if_item_exists 쿼리는 다음과 같아야 합니다.
# ... (이전 코드 동일) ...
item_data = item.model_dump(exclude_unset=True)
db_item.sqlmodel_update(item_data)
check_if_item_exists = session.exec(
select(Item)
.where(Item.title == item.title)
.where(Item.foreign_key_id == item.foreign_key_id)
.where(Item.id != item_id) # <<-- 이 줄을 추가합니다.
).first()
if check_if_item_exists:
raise HTTPException(
status_code=409, detail="Dit item bestaat al."
)
session.add(db_item)
session.commit()
session.refresh(db_item)
return db_item
이 변경사항으로 인해 select 쿼리는 이제 item.title과 item.foreign_key_id가 일치하면서도 item_id가 아닌 다른 항목을 찾게 됩니다. 만약 이러한 '다른' 항목이 존재한다면, 진정한 의미의 중복이므로 409 Conflict 에러를 올바르게 발생시킬 것입니다.
4. 향후 예방을 위한 팁
이와 같은 유형의 에러는 데이터 업데이트 로직에서 흔히 발생할 수 있으므로, 몇 가지 예방 팁을 고려하는 것이 좋습니다.
데이터베이스 레벨의 고유 제약 조건 활용
애플리케이션 로직에서 중복 검사를 하는 대신, 데이터베이스 자체에 고유 제약 조건(Unique Constraint)을 설정하는 것이 가장 강력하고 효율적인 방법입니다. 예를 들어, SQLModel 모델에 UniqueConstraint를 추가하여 (title, foreign_key_id) 조합의 고유성을 보장할 수 있습니다.
from sqlmodel import SQLModel, Field, UniqueConstraint
from typing import Optional
from uuid import UUID, uuid4
class Item(SQLModel, table=True):
id: Optional[UUID] = Field(default_factory=uuid4, primary_key=True)
title: str
foreign_key_id: UUID
__table_args__ = (UniqueConstraint("title", "foreign_key_id", name="uq_title_foreign_key_id"),)
이렇게 하면 데이터베이스 레벨에서 중복 삽입/업데이트 시 에러가 발생하며, 이를 FastAPI 예외 핸들러에서 적절히 처리할 수 있습니다.
조건부 업데이트 로직
업데이트 요청으로 들어온 item.title이 db_item.title과 실제로 다른 경우에만 중복 검사를 수행하도록 로직을 최적화할 수 있습니다. 불필요한 데이터베이스 쿼리를 줄일 수 있습니다.
# ... (생략) ...
db_item_before_update = db_item.model_copy(deep=True) # 업데이트 전 상태를 복사
item_data = item.model_dump(exclude_unset=True)
db_item.sqlmodel_update(item_data)
# 제목이 변경되었거나, foreign_key_id가 변경되었을 때만 중복 검사
if (db_item.title != db_item_before_update.title or
db_item.foreign_key_id != db_item_before_update.foreign_key_id):
check_if_item_exists = session.exec(
select(Item)
.where(Item.title == db_item.title)
.where(Item.foreign_key_id == db_item.foreign_key_id)
.where(Item.id != item_id)
).first()
if check_if_item_exists:
raise HTTPException(
status_code=409, detail="Dit item bestaat al."
)
# ... (생략) ...
이 코드는 db_item.title에 item.title의 값이 아직 반영되지 않은 상태에서 비교하는 것을 의미합니다. 위에서 설명한 것처럼 db_item.sqlmodel_update(item_data) 이후 db_item은 이미 변경된 값을 가지고 있기 때문에, 비교 로직은 db_item_before_update.title (원래 값)과 item.title (새로운 요청 값)을 비교하는 것이 더 정확합니다.
# ... (생략) ...
# 요청 본문에 title이나 foreign_key_id가 포함되어 있고, 기존 값과 다를 때만 검사
new_title = item_data.get("title")
new_foreign_key_id = item_data.get("foreign_key_id")
if (new_title and new_title != db_item.title) or \
(new_foreign_key_id and new_foreign_key_id != db_item.foreign_key_id):
# 중복 검사 쿼리에서는 변경될 값을 사용해야 함
check_title = new_title if new_title else db_item.title
check_foreign_key_id = new_foreign_key_id if new_foreign_key_id else db_item.foreign_key_id
check_if_item_exists = session.exec(
select(Item)
.where(Item.title == check_title)
.where(Item.foreign_key_id == check_foreign_key_id)
.where(Item.id != item_id)
).first()
if check_if_item_exists:
raise HTTPException(
status_code=409, detail="Dit item bestaat al."
)
# 중복 검사 후, 실제 업데이트 적용 (이미 SQLModel 업데이트는 되어 있음)
session.add(db_item)
session.commit()
session.refresh(db_item)
return db_item
이 방식은 요청된 변경 사항이 실제로 기존 데이터와 달라질 때만 중복 검사를 수행하여 효율성을 높입니다.
테스트 코드 작성
이러한 엣지 케이스를 잡아내기 위해 단위 및 통합 테스트를 철저히 작성하는 것이 중요합니다. 특히 다음 시나리오에 대한 테스트를 포함해야 합니다.
- 제목만 변경 (중복 없는 경우)
- 제목만 변경 (다른 항목과 중복되는 경우)
foreign_key_id만 변경 (중복 없는 경우)foreign_key_id만 변경 (다른 항목과 중복되는 경우)- 제목과
foreign_key_id모두 변경 (중복 없는 경우) - 제목과
foreign_key_id모두 변경 (다른 항목과 중복되는 경우) - 제목이나
foreign_key_id가 변경되지 않은 경우
데이터 업데이트 로직은 API의 핵심 기능 중 하나이므로, 견고하고 정확하게 구현하는 것이 중요합니다. 이 글이 유사한 문제를 겪는 개발자분들께 큰 도움이 되기를 바랍니다!
![[문제 해결] [Celery] 여러 태스크 결과 병합 및 데이터 중복 제거, 벌크 쓰기 최적화 [문제 해결] [Celery] 여러 태스크 결과 병합 및 데이터 중복 제거, 벌크 쓰기 최적화](https://dev-error.com/wp-content/plugins/contextual-related-posts/default.png)