우리 시스템에는 사용자의 사용 경험을 개선하기 위한 Event 테이블이 있다. 이 테이블은 사용자가 최근에 어떤 사전(Dictionary)과 해석(Interpretation)을 사용했는지 그 기록을 JSON 필드 안에 저장한다. 그런데 여기서 데이터 무결성 문제가 발생한다. 참조 정보가 일반적인 칼럼이 아닌 JSON 필드 내부에 있기 때문에, 데이터베이스의 기본 기능인 외래 키(Foreign Key) 제약조건을 설정할 수 없다. 즉, 데이터베이스 레벨에서는 사전이나 해석이 삭제되더라도 Event 테이블이 그 사실을 알 방법이 없다. 만약 사용자의 요청대로 사전 데이터를 실제로 삭제해 버린다면, Event 테이블에 남아있는 참조 정보는 더 이상 존재하지 않는 대상을 가리키게 된다.
이처럼 데이터베이스 레벨에서 무결성을 보장할 수 없을 때, 우리는 애플리케이션, 즉 ORM 레벨에서 해법을 찾아야 한다. 그래서 우리는 데이터를 실제로 지우지 않고 '삭제된 척'하는 전략, Soft Delete(소프트 삭제)를 사용하기로 했다. 이는 파일을 휴지통에 버리는 것과 비슷하다. 파일은 눈앞에서 사라지지만, 실제로는 휴지통 안에 보관되어 있어 언제든 복원할 수 있다.
데이터베이스에서 Soft Delete는 새로운 필드(deleted_at)을 만들고 여기에 삭제 날짜를 적어 삭제 여부를 확인하는 방식으로 구현하는 것이 가장 일반적이다. Soft Delete를 제대로 구현하려면 '지우는 작업'과 '읽는 작업'이라는 두 가지 측면을 모두 고민해야 한다.
삭제하기 구현법
'삭제' 요청이 들어왔을 때, 실제로 DELETE 쿼리를 보내는 대신 deleted_at 칼럼에 현재 시간을 기록하는 UPDATE 쿼리를 보내야 한다. 여기에는 두 가지 방법이 있다.
첫 번째는 단순한 방법이다. 코드 상에서 session.delete(object)를 호출하는 모든 부분을 찾아 object.deleted_at = datetime.now() 로 직접 수정하는 것이다.
두 번째는 고급 방법이다. Mixin(믹스인)과 이벤트 리스너(Event Listener)를 활용해 이 과정을 자동화하는 것이다. 이 방법은 개발자의 실수를 차단하고, 코드의 일관성을 유지할 수 있게 도움을 준다. 이 글에서는 두 번째 방법을 다루고 그 한계를 살펴본다.
먼저, Soft Delete에 필요한 공통 필드와 메서드를 가진 SoftDeleteMixin을 만든다. 그리고 SQLAlchemy의 이벤트 시스템을 활용해, 데이터베이스에 변경사항을 반영하기 직전(before_flush)의 순간을 가로챈다.
이 이벤트 리스너는 데이터베이스로 가는 길목을 지키는 문지기와 같다. 이 문지기는 삭제 목록에 포함된 모든 객체를 검사하여, 만약 그 객체가 SoftDeleteMixin의 자손이라면 물리적 삭제를 막고 대신 soft_delete() 메서드를 호출하여 '삭제된 척' 상태로 바꿔준다.
from sqlalchemy import event, Column, DateTime
from datetime import datetime, UTC
class SoftDeleteMixin:
deleted_at: Mapped[datetime | None] = mapped_column(DateTime(timezone=True), nullable=True)
def soft_delete(self):
self.deleted_at = datetime.now(UTC)
Python
복사
SQLAlchemy 2.0
from sqlalchemy.orm.session import Session
from sqlalchemy.orm.context import ORMFlushContext
from sqlalchemy.orm.state import InstanceState
@event.listens_for(Session, 'before_flush')
def receive_before_flush(session: Session, flush_context: ORMFlushContext, instances: list[InstanceState]):
for instance in session.deleted:
if isinstance(instance, SoftDeleteMixin):
instance.soft_delete()
session.add(instance)
Python
복사
SQLAlchemy 2.0
이제 개발자는 Soft Delete라는 경우를 고려하지 않고 session.delete(my_dictionary)를 호출할 수 있다.
삭제된 데이터를 제외하고 읽는 방법
이제 '삭제된 척'하는 데이터는 사용자의 눈에 보이지 않도록 숨겨야 한다. 이 또한 두 가지 방법이 있다.
첫 번째는 삭제하기를 위한 첫 번째 방법에 대응되는 단순한 방법이다. 데이터를 조회하는 모든 메서드에 deleted_at이 NULL인 데이터만 가져오도록 필터링 조건을 일일이 추가하는 것이다.
두 번째는 고급 방법이다. Query 객체에 대한 이벤트 리스너를 사용하여, 모든 SELECT 쿼리에 자동으로 필터링 조건을 추가하는 것이다. 여기서도 우리는 두 번째 방법을 알아본다. 이 리스너는 모든 SELECT 쿼리에 deleted_at IS NULL 이라는 조건을 자동으로 붙여 준다. 개발자가 깜빡하고 필터링 조건을 빼먹는 실수를 할 가능성을 없애준다.
from sqlalchemy.orm.query import ORMExecuteState
@event.listens_for(Session, 'do_orm_execute')
def soft_delete_criteria(orm_execute_state: ORMExecuteState):
if orm_execute_state.execution_options.get('include_deleted', False):
return
# SELECT 문일 경우에만 필터링 로직 적용
if orm_execute_state.is_select:
for entity in orm_execute_state.statement.column_descriptions:
orm_entity = entity['entity']
if orm_entity and issubclass(orm_entity, SoftDeleteMixin):
orm_execute_state.statement = orm_execute_state.statement.where(orm_entity.deleted_at.is_(None))
Python
복사
SQLAlchemy 2.0
이처럼 Mixin과 이벤트 리스너를 활용하면, 개발자는 Soft Delete의 내부 구현을 신경 쓰지 않고 평소처럼 session.delete()와 session.query()를 사용할 수 있다.
참고: 아래와 같은 모양의 코드는 SQLAlchemy 1.0 전용 코드다.
부모-자식 관계에서의 연쇄 Soft Delete 설정
고오급 방법을 사용하면 특히 부모-자식 관계 테이블들 사이의 Soft Delete를 우아하게 처리할 수 있다는 장점이 있다. 부모인 Dictionary를 삭제할 때, 자식인 Interpretation들도 잊지 않고 함께 Soft Delete 처리해야 실수가 없다. 이 자동 연쇄 반응은 어떻게 설정할까?
정답은 ORM 레벨의 relationship에 cascade 옵션을 명시하는 것이다. 이는 데이터베이스 레벨의 ON DELETE CASCADE와는 별개로, SQLAlchemy Session 내부에서만 작동하는 애플리케이션 레벨의 규칙이다.
from sqlalchemy.orm import DeclarativeBase, Mapped, mapped_column, relationship
class Dictionary(DeclarativeBase, SoftDeleteMixin):
interpretations: Mapped[List["Interpretation"]] = relationship(
back_populates="dictionary",
cascade="all, delete-orphan",
passive_deletes=True,
)
class Interpretation(DeclarativeBase, SoftDeleteMixin):
dictionary_id: Mapped[int] = mapped_column(ForeignKey("dictionary.id", ondelete="CASCADE"))
dictionary: Mapped["Dictionary"] = relationship(back_populates="interpretations")
Python
복사
SQLAlchemy 2.0
여기서 cascade="all, delete-orphan"은 SQLAlchemy의 Session에게 다음과 같이 지시한다. "만약 Dictionary 객체가 Session 내에서 삭제(delete)되거나 부모-자식 관계가 끊어지면(orphan), 그 변경사항을 자식인 Interpretation 객체들에게도 모두 전파(cascade)하라." 이 ORM 레벨의 지시와 우리가 만든 이벤트 리스너가 만나면 다음과 같은 일이 벌어진다.
1.
개발자가 session.delete(my_dictionary)를 호출한다.
2.
SQLAlchemy Session은 cascade 규칙에 따라, my_dictionary에 속한 모든 interpretation 객체들도 삭제 목록에 추가한다.
3.
before_flush 이벤트가 발생하고, 우리의 문지기(이벤트 리스너)가 등장한다.
4.
문지기는 삭제 목록에 있는 Dictionary 객체와 모든 Interpretation 객체들을 발견하고, 이들 모두를 물리적으로 삭제하는 대신 deleted_at 필드를 업데이트하는 Soft Delete로 처리한다.
결과적으로, 단 한 번의 delete 호출로 부모와 자식 모두가 누락 없이 안전하게 Soft Delete 자동화를 ORM 레벨에서 구현할 수 있다. ForeignKey("dictionary.id", ondelete="CASCADE") 옵션 덕분에 시스템 권한으로 데이터를 제어하기 위해 DB에 직접 DELETE 쿼리를 보내는 경우에도 무결성을 보장할 수 있다.
마법의 한계
우리가 만든 이벤트 리스너 기반의 Soft Delete 시스템은 대부분의 경우 훌륭하게 작동한다. 일반적인 조회 쿼리에서는 '삭제된' 데이터를 마법처럼 자동으로 걸러내 주어, 개발자는 Soft Delete의 존재를 잊고 코드를 작성할 수 있다.
하지만 당연히 이 시스템도 완벽하지 않다. 특히 SQLAlchemy의 Eager Loading(즉시 로딩) 전략, 그중에서도 JoinedLoad를 사용하는 경우 숨겨진 함정이 드러난다. 문제 상황은 다음과 같다. Interpretation을 조회하면서, 연관된 Dictionary 객체를 함께 가져오기 위해 joinedload 옵션을 사용한다고 가정해 보자.
stmt = select(Interpretation).options(joinedload(Interpretation.dictionary))
results = session.execute(stmt).scalars().all()
Python
복사
SQLAlchemy 2.0
SQLAlchemy 1.0
이 코드가 실행될 때, 우리가 만든 '읽기' 이벤트 리스너는 Interpretation이 주된 조회 대상임을 인지하고 WHERE interpretation.deleted_at IS NULL 조건을 쿼리에 자동으로 추가한다. 여기까지는 완벽하다. 우리는 삭제되지 않은 Interpretation 목록을 정확히 얻을 수 있다.
하지만 문제는 JOIN되는 Dictionary 테이블에 있다. 이벤트 리스너는 주된 SELECT 대상이 아닌, JOIN되는 대상의 필터링까지는 책임지지 않는다. 따라서 쿼리는 interpretation.dictionary_id를 기준으로 Dictionary 테이블과 LEFT JOIN을 수행하는데, 이때 상대방 Dictionary가 Soft Delete 되었는지는 검사하지 않는다. 결과적으로, 삭제되지 않은 Interpretation 객체에 이미 삭제 처리된 Dictionary 객체가 붙어서 로드되는 상황이 발생할 수 있다.
물론 이 문제까지 우회하기 위해 relationship을 정의할 때 primaryjoin 조건을 사용하여, 관계를 맺는 시점 자체에서 Soft Delete된 대상을 원천적으로 제외하는 방법도 있다.
dictionary: Mapped["Dictionary"] = relationship(
primaryjoin=and_("Interpretation.dictionary_id == Dictionary.id", Dictionary.deleted_at.is_(None)),
back_populates="interpretations"
)
Python
복사
SQLAlchemy 2.0
이 방법은 관계를 탐색하는 모든 경우에 필터링이 적용되므로 매우 강력하지만, 의도치 않은 사이드 이펙트를 유발할 수 있어 신중하게 사용해야 한다. 결국 끝까지 우아한 방법은 없는 것 같다. joinedload 대신 명시적인 join을 사용하고, JOIN하는 대상 테이블에 대한 필터링 조건을 직접 걸어주는 방식을 사용해야 한다.
from sqlalchemy.orm import contains_eager
# JOIN을 통해 Dictionary의 상태까지 직접 확인
stmt = (
select(Interpretation)
.join(Interpretation.dictionary)
.filter(Dictionary.deleted_at.is_(None))
.options(contains_eager(Interpretation.dictionary))
)
results = session.execute(stmt).scalars().all()
Python
복사
SQLAlchemy 2.0
SQLAlchemy 1.0
이렇게 된다면, 개발자들도 Soft Delete의 존재에 대해 명확히 인지하고 있어야 한다. 결국 첫 번째 방법도 충분히 명시적이고 우아한 방법이다.
parse me : 언젠가 이 글에 쓰이면 좋을 것 같은 재료을 보관해 두는 영역입니다.
1.
None
from : 과거의 어떤 원자적 생각이 이 생각을 만들었는지 연결하고 설명합니다.
1.
•
SQLAlchemy
2.
•
현재는 포함되어있지 않지만, soft delete가 필요한 경우 이 글의 스니펫들을 프롬프트로 사용할 수 있다.
supplementary : 어떤 새로운 생각이 이 문서에 작성된 생각을 뒷받침하는지 연결합니다.
opposite : 어떤 새로운 생각이 이 문서에 작성된 생각과 대조되는지 연결합니다.
1.
None
to : 이 문서에 작성된 생각이 어떤 생각으로 발전되거나 이어지는지를 작성하는 영역입니다.
1.
None
ref : 생각에 참고한 자료입니다.
1.
None