Soft delete in SQLAlchemy: mixins and events

Capsule overview

You already have the base pattern in PostgreSQL: a deleted_at TIMESTAMPTZ NULL column + a partial index. Now comes the operational problem: in a codebase with 50 queries and a team of 10 people, how do you guarantee every new query includes WHERE deleted_at IS NULL? The answer "everyone always remembers" doesn't scale. What scales is automating the filter at the ORM level.

SQLAlchemy 2.0 offers three mechanisms for this: a before_compile event listener that injects the filter automatically, a custom query class that replaces the query factory with one that has the filter hardcoded, and a mixin with @declared_attr that only adds the column and the helper methods without touching the query. Each one has trade-offs. This capsule implements them all, compares them with concrete criteria, and recommends mixin + before_compile event listener as the default combination. It's the decision you'll take to the module project (capsule 08) and to the capstone project (module 8 of the guide).

By the end you'll have runnable code for all three approaches, criteria to defend your choice in a code review, and a clean escape hatch for the cases where you do need to see deleted records (audit, recovery, analytics). It's the module's most important technical capsule.

The problem: the forgotten filter in production

Picture this sequence. Your app has tasks with soft delete. The existing queries include WHERE deleted_at IS NULL. Code reviews are rigorous. Everything is fine.

A teammate implements a new endpoint:

@app.get("/tasks/recent")
async def recent_tasks(db: AsyncSession = Depends(get_session)):
    result = await db.execute(
        select(Task).order_by(Task.created_at.desc()).limit(20)
    )
    return result.scalars().all()

QA passes. Tests pass. Deploy. Three days later, support gets a ticket: "I'm seeing tasks I deleted in my recent feed." The bug is obvious in hindsight: where(Task.deleted_at.is_(None)) is missing. But nobody caught it in the code review because the endpoint was trivial and "nobody would think to write a query without that filter."

This bug is semantic, not a performance one. The SQL plan is correct, the query returns data, the types are right. It's just that the semantics are wrong. Unit tests typically don't catch it (who writes a specific test for "verify the endpoint doesn't return soft-deleted rows"?).

The solution isn't "more rigorous code reviews." The solution is to remove the possibility: make it impossible to write a query against Task without the filter, unless you explicitly ask to see the deleted ones. That's what the automatic mechanism does.

Mental model: the filter as a layer of abstraction

Think of soft delete like authentication. Nobody on your team writes if not user.is_authenticated: return 401 in every endpoint — you use a decorator or middleware that does it automatically. If someone has an endpoint that genuinely has to be public, they mark it explicitly. The default rule is "authenticated"; the opt-out is explicit.

Soft delete should work the same way. By default, every query excludes soft-deleted rows. If someone needs to see them (an auditor, a recovery flow), they opt in explicitly. The operational difference is enormous: a new teammate can write queries without remembering the filter and the system imposes the correct semantics on them.

The three options in SQLAlchemy 2.0

Let's look at them in order of increasing inversion of control: from the manual approach with helpers only, to the automatic approach that intercepts every query.

Option 1: Mixin with @declared_attr (no automatic filtering)

The most basic mixin only adds the column and the helper methods. It doesn't automate anything about the filter.

# app/models/mixins.py
from datetime import datetime, timezone
from typing import ClassVar

from sqlalchemy import DateTime
from sqlalchemy.orm import Mapped, declared_attr, mapped_column


class SoftDeleteMixin:
    """
    Mixin that adds a `deleted_at` column and helpers to a model.
    Does NOT automate query filtering: every query has to explicitly
    include `where(Model.deleted_at.is_(None))`.
    """

    @declared_attr
    def deleted_at(cls) -> Mapped[datetime | None]:
        return mapped_column(
            DateTime(timezone=True),
            nullable=True,
            default=None,
            index=False,  # the partial index is defined at the table level
        )

    def soft_delete(self) -> None:
        """Marks the entity as soft-deleted."""
        self.deleted_at = datetime.now(timezone.utc)

    def restore(self) -> None:
        """Restores a soft-deleted entity."""
        self.deleted_at = None

    @property
    def is_deleted(self) -> bool:
        return self.deleted_at is not None

Usage:

# app/models/task.py
from sqlalchemy import BigInteger, Index, String, DateTime, func
from sqlalchemy.orm import DeclarativeBase, Mapped, mapped_column

from app.models.mixins import SoftDeleteMixin


class Base(DeclarativeBase):
    pass


class Task(Base, SoftDeleteMixin):
    __tablename__ = "tasks"

    id: Mapped[int] = mapped_column(BigInteger, primary_key=True, autoincrement=True)
    author_id: Mapped[int] = mapped_column(BigInteger, nullable=False, index=True)
    title: Mapped[str] = mapped_column(String(200), nullable=False)
    created_at: Mapped[datetime] = mapped_column(
        DateTime(timezone=True),
        nullable=False,
        server_default=func.now(),
    )

    # Partial index: the canonical pattern
    __table_args__ = (
        Index(
            "idx_tasks_author_created_active",
            "author_id",
            "created_at",
            postgresql_where=text("deleted_at IS NULL"),
        ),
    )

Every query is still manual:

# Delete
task = await session.get(Task, task_id)
task.soft_delete()
await session.commit()

# List (manual filter is mandatory)
result = await session.execute(
    select(Task).where(Task.deleted_at.is_(None))
)

Trade-offs:

When to use it: very small apps, scripts, prototypes, teams of 1-2 people with high discipline. NOT recommended for multi-team production.

Option 2: Custom query class

SQLAlchemy 1.x had first-class support for query classes; SQLAlchemy 2.0 discouraged this pattern in favor of event listeners. Even so, some people implement it with a wrapper around select(). The idea: a select_active(Model) factory function that returns a Select with the filter already applied.

# app/db/queries.py
from typing import TypeVar
from sqlalchemy import Select, select

from app.models.mixins import SoftDeleteMixin

ModelT = TypeVar("ModelT", bound=SoftDeleteMixin)


def select_active(model: type[ModelT]) -> Select[tuple[ModelT]]:
    """
    Factory that returns a Select with the soft delete filter applied.
    Usage: `select_active(Task).where(Task.author_id == 42)`.
    """
    return select(model).where(model.deleted_at.is_(None))


def select_with_deleted(model: type[ModelT]) -> Select[tuple[ModelT]]:
    """Factory that does NOT apply the filter. For audit, recovery, analytics."""
    return select(model)

Usage:

# List (uses the factory, the filter is inside)
result = await session.execute(
    select_active(Task).where(Task.author_id == 42)
)

# Audit (a factory that sees deleted rows)
result = await session.execute(
    select_with_deleted(Task).where(Task.author_id == 42)
)

Trade-offs:

When to use it: teams that prefer explicitness and accept the discipline of "always use the factory." The forgotten filter bug is still possible if someone writes select(Task) directly. Convention with no enforcement.

Option 3: before_compile event listener (automatic interception)

SQLAlchemy offers events at the Session and Query level. The do_orm_execute event (the modern successor to before_compile in SQLAlchemy 2.0) fires before executing any ORM statement, giving you the chance to modify it.

# app/db/soft_delete_filter.py
from typing import Any

from sqlalchemy import event, Select
from sqlalchemy.orm import Session, with_loader_criteria

from app.models.mixins import SoftDeleteMixin


def install_soft_delete_filter(target_session_class: type[Session]) -> None:
    """
    Registers the event listener that injects `WHERE deleted_at IS NULL`
    into every ORM query that touches models with SoftDeleteMixin.

    To skip the filter on a specific query:
        result = session.execute(
            select(Task).execution_options(include_deleted=True)
        )
    """

    @event.listens_for(target_session_class, "do_orm_execute")
    def _add_soft_delete_filter(orm_execute_state) -> None:
        # If the query explicitly asked to see deleted rows, don't inject
        if orm_execute_state.execution_options.get("include_deleted", False):
            return

        # If it isn't a SELECT, don't inject (UPDATE/DELETE don't need the filter)
        if not orm_execute_state.is_select:
            return

        # Inject the filter through with_loader_criteria.
        # This applies the WHERE to ANY SoftDeleteMixin entity that shows up
        # in the query, including joins and relationships.
        orm_execute_state.statement = orm_execute_state.statement.options(
            with_loader_criteria(
                SoftDeleteMixin,
                lambda cls: cls.deleted_at.is_(None),
                include_aliases=True,
            )
        )

Activation in the app's setup:

# app/db.py
from sqlalchemy.ext.asyncio import (
    AsyncSession,
    async_sessionmaker,
    create_async_engine,
)

from app.db.soft_delete_filter import install_soft_delete_filter

DATABASE_URL = "postgresql+asyncpg://postgres:postgres@localhost:5432/softdelete_demo"

engine = create_async_engine(DATABASE_URL, echo=False, pool_pre_ping=True)

AsyncSessionLocal = async_sessionmaker(
    engine,
    class_=AsyncSession,
    expire_on_commit=False,
)

# Register the automatic filter on the session class
install_soft_delete_filter(AsyncSession)


async def get_session():
    async with AsyncSessionLocal() as session:
        yield session

Usage:

# List (automatic filter applied, without writing it)
result = await session.execute(select(Task).where(Task.author_id == 42))
# Actual SQL executed:
#   SELECT * FROM tasks
#   WHERE tasks.author_id = 42
#     AND tasks.deleted_at IS NULL

# Audit (explicit escape hatch)
result = await session.execute(
    select(Task)
    .where(Task.author_id == 42)
    .execution_options(include_deleted=True)
)
# Actual SQL executed:
#   SELECT * FROM tasks WHERE tasks.author_id = 42

# JOIN (the filter is applied to BOTH entities if both are SoftDelete)
result = await session.execute(
    select(Task, Comment).join(Comment, Comment.task_id == Task.id)
)
# Actual SQL executed:
#   SELECT * FROM tasks
#   JOIN comments ON comments.task_id = tasks.id
#   WHERE tasks.deleted_at IS NULL
#     AND comments.deleted_at IS NULL

Trade-offs:

When to use it: real production, teams of >2 people, a codebase with >10 queries using soft delete. It's the approach SQLAlchemy 2.0 endorses for this use case.

Recommendation: mixin + event listener as the default

The combination you'll take to the module project (capsule 08) and to the capstone project (module 8) is:

• Mixin (SoftDeleteMixin) to add the column and the helper methods in a reusable way.
• do_orm_execute event listener to automate the filter across all SELECT queries.
• An explicit escape hatch (execution_options(include_deleted=True)) for audit/recovery.

Why this combination wins:

1. The mixin alone is insufficient: it automates the column but leaves the filter up to the developer. The forgotten filter bug is still possible.

2. The custom query class is convention with no enforcement: it works if everyone uses the factory, but a direct select(Task) (perfectly valid syntactically) silently bypasses the filter.

3. The event listener alone is hard to maintain: without the mixin, there's no way to know which entities have soft delete. The mixin is the explicit marker: "this model has deleted_at and the filter has to be applied to it."

4. Combined, the system is safe by default: the developer writes queries without thinking about the filter and the system guarantees the correct semantics. Opting out requires conscious action and stays visible in code review.

5. It's what SQLAlchemy 2.0 documents: the official "soft delete" example in the with_loader_criteria documentation uses exactly this pattern.

Complete final implementation

# app/models/mixins.py
from datetime import datetime, timezone
from sqlalchemy import DateTime
from sqlalchemy.orm import Mapped, declared_attr, mapped_column


class SoftDeleteMixin:
    """
    Standard mixin for soft delete.

    Adds:
    - A `deleted_at TIMESTAMPTZ NULL` column
    - A `soft_delete()` method: marks as deleted
    - A `restore()` method: restores
    - An `is_deleted` property: bool

    For the automatic filtering to work, you have to register
    `install_soft_delete_filter(SessionClass)` in the app's setup.
    """

    @declared_attr
    def deleted_at(cls) -> Mapped[datetime | None]:
        return mapped_column(
            DateTime(timezone=True),
            nullable=True,
            default=None,
        )

    def soft_delete(self) -> None:
        self.deleted_at = datetime.now(timezone.utc)

    def restore(self) -> None:
        self.deleted_at = None

    @property
    def is_deleted(self) -> bool:
        return self.deleted_at is not None

# app/db/soft_delete_filter.py
from sqlalchemy import event
from sqlalchemy.orm import Session, with_loader_criteria

from app.models.mixins import SoftDeleteMixin


def install_soft_delete_filter(target_session_class: type[Session]) -> None:
    """
    Registers the listener that automatically injects `WHERE deleted_at IS NULL`
    into SELECT queries that touch SoftDeleteMixin entities.

    Escape hatch:
        session.execute(select(Task).execution_options(include_deleted=True))
    """

    @event.listens_for(target_session_class, "do_orm_execute")
    def _filter_soft_deleted(orm_execute_state) -> None:
        if orm_execute_state.execution_options.get("include_deleted", False):
            return
        if not orm_execute_state.is_select:
            return

        orm_execute_state.statement = orm_execute_state.statement.options(
            with_loader_criteria(
                SoftDeleteMixin,
                lambda cls: cls.deleted_at.is_(None),
                include_aliases=True,
            )
        )

# app/models/task.py
from datetime import datetime
from sqlalchemy import BigInteger, Index, String, DateTime, func, text
from sqlalchemy.orm import DeclarativeBase, Mapped, mapped_column

from app.models.mixins import SoftDeleteMixin


class Base(DeclarativeBase):
    pass


class Task(Base, SoftDeleteMixin):
    __tablename__ = "tasks"

    id: Mapped[int] = mapped_column(BigInteger, primary_key=True, autoincrement=True)
    author_id: Mapped[int] = mapped_column(BigInteger, nullable=False)
    title: Mapped[str] = mapped_column(String(200), nullable=False)
    created_at: Mapped[datetime] = mapped_column(
        DateTime(timezone=True),
        nullable=False,
        server_default=func.now(),
    )

    __table_args__ = (
        Index(
            "idx_tasks_author_created_active",
            "author_id",
            "created_at",
            postgresql_where=text("deleted_at IS NULL"),
        ),
    )

# app/db.py
from sqlalchemy.ext.asyncio import (
    AsyncSession,
    async_sessionmaker,
    create_async_engine,
)

from app.db.soft_delete_filter import install_soft_delete_filter

DATABASE_URL = "postgresql+asyncpg://postgres:postgres@localhost:5432/softdelete_demo"

engine = create_async_engine(DATABASE_URL, echo=False, pool_pre_ping=True)

AsyncSessionLocal = async_sessionmaker(
    engine, class_=AsyncSession, expire_on_commit=False
)

install_soft_delete_filter(AsyncSession)


async def get_session():
    async with AsyncSessionLocal() as session:
        yield session

End-to-end verification

# scripts/verify_soft_delete.py
import asyncio

from sqlalchemy import select

from app.db import AsyncSessionLocal, engine
from app.models.task import Base, Task


async def main() -> None:
    async with engine.begin() as conn:
        await conn.run_sync(Base.metadata.drop_all)
        await conn.run_sync(Base.metadata.create_all)

    async with AsyncSessionLocal() as session:
        # Create 3 tasks: 2 active, 1 deleted
        t1 = Task(author_id=1, title="Active 1")
        t2 = Task(author_id=1, title="Active 2")
        t3 = Task(author_id=1, title="Deleted")
        session.add_all([t1, t2, t3])
        await session.commit()

        # Soft-delete one
        t3.soft_delete()
        await session.commit()

        # Default query: automatic filter
        result = await session.execute(select(Task).where(Task.author_id == 1))
        active = result.scalars().all()
        print(f"Default query returned: {[t.title for t in active]}")
        # Expected: ['Active 1', 'Active 2']

        # Escape hatch: see them all
        result = await session.execute(
            select(Task)
            .where(Task.author_id == 1)
            .execution_options(include_deleted=True)
        )
        all_tasks = result.scalars().all()
        print(f"Include deleted query returned: {[t.title for t in all_tasks]}")
        # Expected: ['Active 1', 'Active 2', 'Deleted']


if __name__ == "__main__":
    asyncio.run(main())

python scripts/verify_soft_delete.py
# Default query returned: ['Active 1', 'Active 2']
# Include deleted query returned: ['Active 1', 'Active 2', 'Deleted']

The filter worked automatically without writing where(Task.deleted_at.is_(None)) in the first query. And the escape hatch worked when you asked for it explicitly.

Why does this automation matter in real work?

1. It's the only approach that scales to teams. A developer who's new on the team can write queries without knowing about the soft delete pattern. The system imposes the correct semantics on them. Without this, soft delete is growing technical debt: every new endpoint is an opportunity for a bug.

2. The escape hatch is explicit and reviewable. When someone writes execution_options(include_deleted=True), it's visible in the PR diff. The reviewer can ask "why do you need to see deleted rows here?". Without this, the queries that see deleted rows are indistinguishable from the ones that filter (both look like select(Task)).

3. It works with relationships and joins automatically. with_loader_criteria applies to any alias of the model in the query. That includes joins and eager loading (selectinload, joinedload). Without the listener, every relationship would require its own manual filter.

4. It's the officially documented pattern. SQLAlchemy 2.0 endorses with_loader_criteria for soft delete. You aren't inventing a workaround; you're using the tool designed for the case. That means: maintainable code, future-compatible, and recognizable to any dev who knows SQLAlchemy.

Traps and common mistakes

Mistake 1 (conceptual): assuming the filter applies to UPDATE/DELETE

Symptom: you run await session.execute(update(Task).where(Task.id == 5).values(title="X")) and a teammate asks "shouldn't it also be filtered by deleted_at IS NULL automatically?".

Why it's wrong: the listener as we implemented it only intercepts SELECTs (if not orm_execute_state.is_select: return). UPDATEs and DELETEs aren't filtered because you typically want to operate on any row by id, soft-deleted or not (recovery, for example, needs updates on deleted rows).

How to tell: if you need the "don't update deleted rows" semantics, add the filter explicitly:

await session.execute(
    update(Task)
    .where(Task.id == 5, Task.deleted_at.is_(None))
    .values(title="X")
)

How to fix it: the rule is "the automatic filter applies to SELECTs; mutations require an explicit filter if you need it." It's consistent with the defensive filter on the soft delete UPDATE you saw in capsule 03.

Mistake 2 (practical): with_loader_criteria with include_aliases=False and joins fail

Symptom: with joins or aliases (aliased(Task)), the filter is NOT applied to the alias, only to the main model.

Why it happens: with_loader_criteria by default only applies to the exact model. Aliases and subqueries aren't covered unless you pass include_aliases=True.

How to tell: run a query with a join and aliased:

TaskAlias = aliased(Task)
result = await session.execute(
    select(Task, TaskAlias).join(TaskAlias, TaskAlias.author_id == Task.author_id)
)

If the generated SQL has the filter only on tasks and not on the alias, you're missing include_aliases=True.

How to fix it: make sure the listener has include_aliases=True (as in the recommended implementation). That's why the implementation includes that parameter explicitly.

Mistake 3 (conceptual): thinking the listener intercepts raw SQL (session.execute(text(...)))

Symptom: you run await session.execute(text("SELECT * FROM tasks")) expecting the filter to be applied. It returns deleted rows.

Why it happens: text() runs raw SQL without going through the ORM system. The do_orm_execute listener only intercepts ORM queries, not raw SQL.

How to tell: check the code: if you use text() or engine.execute() directly, the filter doesn't apply. If you use select(Model), it does.

How to fix it: for raw SQL, write the filter by hand or avoid text() when you can. This is a conscious decision: raw SQL is an escape from the ORM abstraction, and with the escape come the manual responsibilities.

Mistake 4 (edge case): the listener gets registered twice (on re-imports or test setup)

Symptom: the filter is applied with a double AND: WHERE deleted_at IS NULL AND deleted_at IS NULL. Functionally correct but ugly and slightly more expensive.

Why it happens: install_soft_delete_filter gets called multiple times during test setup or from uvicorn's hot-reload.

How to tell: enable echo=True on the engine and review the generated SQL. If you see duplicate filters, this is it.

How to fix it: use a single-registration guard:

_REGISTERED = set()

def install_soft_delete_filter(target_session_class):
    if target_session_class in _REGISTERED:
        return
    _REGISTERED.add(target_session_class)

    @event.listens_for(target_session_class, "do_orm_execute")
    def _filter_soft_deleted(orm_execute_state):
        # ... the rest is the same

Mistake 5 (conceptual): a custom query class mixed with an event listener

Symptom: the team used a custom query class (select_active) in parts of the code and an event listener in others. Some queries get filtered twice, others once, nothing breaks but the logic is opaque.

Why it happens: each mechanism is independent. If select_active(Task) already has the filter in the WHERE and then the listener adds it too, the result has the filter duplicated.

How to fix it: pick one mechanism and use it consistently. This capsule's recommendation is the event listener (don't use the custom factory). If you're coming from a codebase with a factory, migrate all the queries and remove the factory.

Exercises

Exercise 1: implement the full pattern in an app from scratch

Set up a FastAPI project with SQLAlchemy 2.0 async and PostgreSQL. Implement the SoftDeleteMixin, the install_soft_delete_filter, a Task entity that uses the mixin, and a GET /tasks endpoint that lists the active ones. Verify with curl that the soft-deleted ones don't show up.

softdelete-fastapi/
├── app/
│   ├── __init__.py
│   ├── db.py
│   ├── models/
│   │   ├── __init__.py
│   │   ├── mixins.py
│   │   └── task.py
│   ├── db/
│   │   ├── __init__.py
│   │   └── soft_delete_filter.py
│   └── main.py
└── seed.py

from datetime import datetime
from typing import List

from fastapi import Depends, FastAPI, HTTPException, status
from pydantic import BaseModel, ConfigDict
from sqlalchemy import select
from sqlalchemy.ext.asyncio import AsyncSession

from app.db import get_session
from app.models.task import Task

app = FastAPI(title="Soft Delete Demo")


class TaskOut(BaseModel):
    model_config = ConfigDict(from_attributes=True)
    id: int
    title: str
    deleted_at: datetime | None


@app.get("/tasks", response_model=List[TaskOut])
async def list_tasks(db: AsyncSession = Depends(get_session)):
    """Lists active tasks (automatic filter)."""
    result = await db.execute(select(Task).order_by(Task.id))
    return result.scalars().all()


@app.get("/tasks/all", response_model=List[TaskOut])
async def list_all_tasks(db: AsyncSession = Depends(get_session)):
    """Lists every task including soft-deleted ones (escape hatch)."""
    result = await db.execute(
        select(Task)
        .order_by(Task.id)
        .execution_options(include_deleted=True)
    )
    return result.scalars().all()


@app.delete("/tasks/{task_id}", status_code=status.HTTP_204_NO_CONTENT)
async def delete_task(task_id: int, db: AsyncSession = Depends(get_session)):
    """Soft delete: marks deleted_at."""
    task = await db.get(Task, task_id)
    if task is None:
        raise HTTPException(status_code=404, detail="Task not found")
    if task.is_deleted:
        raise HTTPException(status_code=410, detail="Task already deleted")
    task.soft_delete()
    await db.commit()

import asyncio
from app.db import AsyncSessionLocal, engine
from app.models.task import Base, Task


async def main():
    async with engine.begin() as conn:
        await conn.run_sync(Base.metadata.drop_all)
        await conn.run_sync(Base.metadata.create_all)

    async with AsyncSessionLocal() as s:
        s.add_all([
            Task(author_id=1, title="Buy milk"),
            Task(author_id=1, title="Walk dog"),
            Task(author_id=1, title="Submit report"),
        ])
        await s.commit()
    print("Seeded 3 tasks")


asyncio.run(main())

python seed.py
uvicorn app.main:app --reload &
curl -s http://localhost:8000/tasks | jq
# 3 tasks listed

curl -X DELETE http://localhost:8000/tasks/2
curl -s http://localhost:8000/tasks | jq
# Only 2 tasks listed (#2 is hidden by the automatic filter)

curl -s http://localhost:8000/tasks/all | jq
# All 3 tasks listed (escape hatch)

SELECT tasks.id, tasks.title, tasks.author_id, tasks.created_at, tasks.deleted_at
FROM tasks
WHERE tasks.deleted_at IS NULL
ORDER BY tasks.id

Exercise 2: compare the three mechanisms against concrete metrics

For each mechanism (mixin alone, query class factory, event listener), answer:

a) How many lines of code does the dev have to write for a new query?

b) Is it possible to forget the filter accidentally?

c) What does the opt-out (explicitly seeing deleted rows) look like?

d) Does it work with joins automatically?

Exercise 3: test the automatic filter and the escape hatch

Write async pytest tests that verify:

a) A default query doesn't return soft-deleted rows. b) A query with execution_options(include_deleted=True) does return them. c) A join between two models with SoftDeleteMixin filters both tables automatically. d) An UPDATE isn't filtered by deleted_at IS NULL (we checked this in common mistake 1).

# tests/test_soft_delete.py
import pytest
from sqlalchemy import select, update
from sqlalchemy.ext.asyncio import AsyncSession

from app.models.task import Task

pytestmark = pytest.mark.asyncio


async def test_filtro_automatico_excluye_soft_deleted(session: AsyncSession):
    t1 = Task(author_id=1, title="Active")
    t2 = Task(author_id=1, title="Deleted")
    session.add_all([t1, t2])
    await session.commit()

    t2.soft_delete()
    await session.commit()

    # Default query: the listener injects WHERE deleted_at IS NULL
    result = await session.execute(select(Task).where(Task.author_id == 1))
    titles = [t.title for t in result.scalars().all()]
    assert titles == ["Active"]


async def test_escape_hatch_devuelve_soft_deleted(session: AsyncSession):
    t1 = Task(author_id=1, title="Active")
    t2 = Task(author_id=1, title="Deleted")
    session.add_all([t1, t2])
    await session.commit()

    t2.soft_delete()
    await session.commit()

    # Explicit escape hatch
    result = await session.execute(
        select(Task)
        .where(Task.author_id == 1)
        .execution_options(include_deleted=True)
    )
    titles = sorted(t.title for t in result.scalars().all())
    assert titles == ["Active", "Deleted"]


async def test_join_filtra_ambas_tablas(session: AsyncSession):
    """
    Assumes Comment also uses SoftDeleteMixin.
    Setup: 1 active task with 2 comments (1 active, 1 deleted).
    """
    from app.models.comment import Comment  # assumed to exist

    task = Task(author_id=1, title="T1")
    session.add(task)
    await session.flush()  # to get task.id

    c1 = Comment(task_id=task.id, body="Active comment")
    c2 = Comment(task_id=task.id, body="Deleted comment")
    session.add_all([c1, c2])
    await session.commit()

    c2.soft_delete()
    await session.commit()

    # JOIN: the filter has to be applied to Task AND Comment
    result = await session.execute(
        select(Task, Comment).join(Comment, Comment.task_id == Task.id)
    )
    rows = result.all()
    assert len(rows) == 1  # only the active comment
    _, comment = rows[0]
    assert comment.body == "Active comment"


async def test_update_no_se_filtra_automaticamente(session: AsyncSession):
    """
    An UPDATE on a soft-deleted row has to work (it isn't filtered).
    This is what enables recovery: UPDATE SET deleted_at = NULL.
    """
    t1 = Task(author_id=1, title="Original")
    session.add(t1)
    await session.commit()

    t1.soft_delete()
    await session.commit()

    # UPDATE on a soft-deleted row: it has to affect the row
    await session.execute(
        update(Task).where(Task.id == t1.id).values(title="Restored")
    )
    await session.commit()

    # Verify with the escape hatch
    result = await session.execute(
        select(Task)
        .where(Task.id == t1.id)
        .execution_options(include_deleted=True)
    )
    task = result.scalar_one()
    assert task.title == "Restored"  # the UPDATE worked
    assert task.is_deleted is True  # it's still soft-deleted

Exercise 4: add the escape hatch to an audit endpoint

Implement an admin endpoint GET /admin/tasks/audit that returns every task (active and soft-deleted) with metadata about when they were deleted. It should only return the soft-deleted ones from the last 90 days.

# app/main.py (extension)
from datetime import datetime, timedelta, timezone
from typing import List

from pydantic import BaseModel
from sqlalchemy import or_

class TaskAuditOut(BaseModel):
    model_config = ConfigDict(from_attributes=True)
    id: int
    title: str
    author_id: int
    created_at: datetime
    deleted_at: datetime | None
    status: str  # "active" or "deleted"


@app.get("/admin/tasks/audit", response_model=List[TaskAuditOut])
async def audit_tasks(db: AsyncSession = Depends(get_session)):
    """
    Audit endpoint: returns active and soft-deleted tasks from the last 90 days.
    Requires the escape hatch to include deleted rows.
    """
    cutoff = datetime.now(timezone.utc) - timedelta(days=90)

    result = await db.execute(
        select(Task)
        .where(
            or_(
                Task.deleted_at.is_(None),  # active
                Task.deleted_at >= cutoff,  # deleted in the last 90 days
            )
        )
        .order_by(Task.deleted_at.desc().nulls_last(), Task.id.desc())
        .execution_options(include_deleted=True)  # CRITICAL: the escape hatch
    )

    tasks = result.scalars().all()
    return [
        TaskAuditOut(
            id=t.id,
            title=t.title,
            author_id=t.author_id,
            created_at=t.created_at,
            deleted_at=t.deleted_at,
            status="deleted" if t.is_deleted else "active",
        )
        for t in tasks
    ]

# Setup: 3 active + 1 soft-deleted today + 1 soft-deleted 200 days ago
curl -s http://localhost:8000/admin/tasks/audit | jq

# Expected output: 4 tasks (3 active + the one deleted today)
# The one deleted 200 days ago does NOT show up (the 90-day cutoff filter)

grep -r "include_deleted" app/
# app/main.py:    .execution_options(include_deleted=True)
# app/admin/recovery.py:    .execution_options(include_deleted=True)

Exercise 5: defend the decision in a code review (real scenario)

A senior teammate proposes using a query class factory (select_active(Task)) instead of an event listener "because it's more explicit and debugger-friendly." It's on you to defend the event listener decision in a PR comment. List 3 concrete arguments with examples.

Summary and next step

In this capsule you learned:

• The forgotten filter is soft delete's most expensive semantic bug. Automation at the ORM level prevents it.
• Three mechanisms in SQLAlchemy 2.0: mixin alone (no filtering), query class factory (convention), do_orm_execute event listener (automatic). Each has concrete trade-offs.
• Default recommendation: mixin (SoftDeleteMixin) + event listener (install_soft_delete_filter) + escape hatch (execution_options(include_deleted=True)). It's what we'll use in the module project and in the capstone project.
• with_loader_criteria with include_aliases=True is SQLAlchemy 2.0's official primitive for this case. It works with joins, relationships, and eager loading with no extra code.
• The listener intercepts only SELECTs, not UPDATEs/DELETEs. This is what allows recovery (UPDATE SET deleted_at = NULL) with no need for a bypass.
• The escape hatch is greppable: grep -r "include_deleted" app/ lists every endpoint that sees deleted rows. It's auditable.

Before moving on you should be able to:

• Implement the full pattern (mixin + listener + escape hatch) in a new FastAPI app in under 30 minutes.
• Justify the choice of event listener vs factory to a senior teammate with 3 concrete arguments.
• Write async pytest tests that verify the automatic filter, the escape hatch, and the behavior in joins.
• Diagnose the forgotten filter bug in a codebase that does NOT use automation (how to find it, how to migrate to the pattern).

Next capsule — Partial indexes for soft deletes. You're going to go deeper into the pattern's other pillar: the partial index. You'll apply what you saw in capsule 03 (the canonical case of partial indexes) with the automatic mechanism from 04, measuring the impact on real queries: how the planner detects the index's implicit filter when the listener injects the WHERE deleted_at IS NULL. It's the bridge to guide #12 (module 3 on advanced indexing) without re-explaining what you already know.

Resources

1. SQLAlchemy 2.0 — with_loader_criteria() — the official reference. The "Adding global WHERE / ON criteria" section describes exactly the soft delete pattern.
2. SQLAlchemy 2.0 — ORM Events: do_orm_execute — the reference for the event our listener uses.
3. SQLAlchemy 2.0 — Mixins and @declared_attr — to understand how the mixin with @declared_attr works.
4. Cultured Systems — "Avoiding the soft delete anti-pattern" — the critical manifesto. Useful for understanding why the forgotten filter is so common.
5. Milan Jovanovic — "Implementing the Soft Delete Pattern" — a comparison of mechanisms in .NET. The reasoning about "global query filters" applies conceptually.
6. Brandur Leach — "Soft deletion probably isn't worth it" — a real case from Stripe. It reinforces why the automation matters when the pattern is applied at scale.
7. django-safedelete documentation — a reference for how Django implements the pattern with managers; useful for comparing approaches across frameworks.
8. SQLAlchemy 2.0 — Versioning Examples — official examples of similar patterns (versioning, audit) that use event listeners.

Module 2 — SQL Patterns for Production APIs Guide

Next capsule: Partial indexes for soft deletes — the canonical case, measurable before-and-after, integration with the automatic filter.