우리 시스템에는 사용자의 사용 경험을 개선하기 위한 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)
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)
이제 개발자는 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))
이처럼 Mixin과 이벤트 리스너를 활용하면, 개발자는 Soft Delete의 내부 구현을 신경 쓰지 않고 평소처럼 session.delete()와 session.query()를 사용할 수 있다.