Anti-patterns: `WHERE deleted_at IS NULL` everywhere

Capsule overview

Even though you automated the filter with the listener (capsule 04) and added the partial index (capsule 05), the soft delete pattern still has gray zones where bugs show up. This capsule is the catalog of anti-patterns you're going to find in real code reviews: the forgotten filter in stats queries (which count deleted rows unintentionally), the silent cascade in foreign keys, JOINs that lose legitimate rows by over-filtering, raw SQL queries that escape the listener, and abuse of include_deleted=True that returns deleted rows to clients by accident.

You're going to learn to recognize them by their signature, understand why they happen, and refactor them into the correct pattern. The goal isn't to memorize a list — it's to develop the reflex of "when I see X, I look at Y," the reflex that turns you into the reviewer the team asks for an opinion when someone introduces soft delete on a new table.

This capsule is defensive by design. If you pass the mental tests it proposes, you can trust that your codebase is protected. If you fail one, you know where to look first.

Anti-pattern #1: the forgotten filter in stats queries

Symptom

Your app exposes a GET /admin/stats endpoint that returns "how many tasks each author has." The SQL looks reasonable:

SELECT author_id, COUNT(*) AS total
FROM tasks
GROUP BY author_id
ORDER BY total DESC;

The report shows author #42 has 800 tasks. The customer opens their team's dashboard and sees 320 tasks. Discrepancy.

Why it happens

The endpoint's query does NOT include deleted_at IS NULL. The automatic listener doesn't filter it either because the endpoint uses text() (raw SQL) to do the grouped COUNT:

result = await session.execute(
    text("SELECT author_id, COUNT(*) AS total FROM tasks GROUP BY author_id")
)

text() skips the ORM system. The do_orm_execute listener doesn't fire. The COUNT counts active + soft-deleted rows.

How to tell

Any stats, dashboard, or report endpoint that:

1. Uses text() or engine.execute() for raw SQL.
2. Uses func.count(), func.sum(), func.avg() over tables with soft delete.
3. Returns numbers the customer sees and compares against another source.

How to fix it

Option A: use the ORM instead of text() (preferred).

from sqlalchemy import func, select

result = await session.execute(
    select(Task.author_id, func.count(Task.id).label("total"))
    .group_by(Task.author_id)
    .order_by(func.count(Task.id).desc())
)
# The listener injects WHERE deleted_at IS NULL automatically.

Option B: if you need raw SQL, add the filter manually.

result = await session.execute(
    text("""
        SELECT author_id, COUNT(*) AS total
        FROM tasks
        WHERE deleted_at IS NULL
        GROUP BY author_id
        ORDER BY total DESC
    """)
)

Option C: if the stat genuinely has to count deleted rows (example: "tasks created historically, regardless of current deletion"), document it explicitly:

# Historical stat: it DOES count deleted tasks. For "live tasks" use /tasks/active-stats.
result = await session.execute(
    text("""
        SELECT author_id, COUNT(*) AS total_lifetime
        FROM tasks
        GROUP BY author_id
    """)
)

The test that catches the bug

async def test_stats_excluye_soft_deleted(session: AsyncSession, client):
    # Create 3 tasks: 2 active + 1 deleted for the same author
    t1 = Task(author_id=1, title="A")
    t2 = Task(author_id=1, title="B")
    t3 = Task(author_id=1, title="C")
    session.add_all([t1, t2, t3])
    await session.commit()
    t3.soft_delete()
    await session.commit()

    res = await client.get("/admin/stats")
    body = res.json()
    author_stats = next(s for s in body if s["author_id"] == 1)
    assert author_stats["total"] == 2  # NOT 3

Anti-pattern #2: the silent cascade in foreign keys

Symptom

Your schema has:

CREATE TABLE comments (
  id BIGSERIAL PRIMARY KEY,
  task_id BIGINT NOT NULL REFERENCES tasks(id) ON DELETE CASCADE,
  body TEXT,
  deleted_at TIMESTAMPTZ NULL
);

The team applied soft delete on tasks and comments. Apparently everything's fine. Until a compliance cleanup job runs:

DELETE FROM tasks WHERE deleted_at < NOW() - INTERVAL '90 days';

The cascade fires and physically removes the comments that pointed at those tasks. The comments with their own deleted_at are lost with no mark, no audit trail, no possible recovery.

Why it happens

ON DELETE CASCADE lives at the schema level. When the app uses soft delete (UPDATE), the cascade doesn't fire. But when someone (a job, a migration, maintenance) runs a real hard delete, the cascade does act. The inconsistency between app-level soft delete and schema-level cascade generates surprises.

How to tell

-- List foreign keys with CASCADE on tables with soft delete
SELECT
  tc.table_name AS child_table,
  kcu.column_name AS child_column,
  ccu.table_name AS parent_table,
  rc.delete_rule
FROM information_schema.referential_constraints rc
JOIN information_schema.table_constraints tc
  ON rc.constraint_name = tc.constraint_name
JOIN information_schema.key_column_usage kcu
  ON tc.constraint_name = kcu.constraint_name
JOIN information_schema.constraint_column_usage ccu
  ON rc.unique_constraint_name = ccu.constraint_name
WHERE rc.delete_rule = 'CASCADE'
  AND ccu.table_name IN (
    SELECT table_name FROM information_schema.columns
    WHERE column_name = 'deleted_at'
  );

Any result is a candidate for auditing.

How to fix it

An explicit decision per relationship. The options:

(1) Change it to RESTRICT or NO ACTION (the most conservative):

ALTER TABLE comments
  DROP CONSTRAINT comments_task_id_fkey,
  ADD CONSTRAINT comments_task_id_fkey
    FOREIGN KEY (task_id) REFERENCES tasks(id) ON DELETE RESTRICT;

Now no hard delete of tasks can run while there are comments pointing at them. You force deleting the children first, which guarantees every operation goes through the app.

(2) Change it to SET NULL if it makes conceptual sense:

ALTER TABLE comments
  ALTER COLUMN task_id DROP NOT NULL;  -- required for SET NULL

ALTER TABLE comments
  DROP CONSTRAINT comments_task_id_fkey,
  ADD CONSTRAINT comments_task_id_fkey
    FOREIGN KEY (task_id) REFERENCES tasks(id) ON DELETE SET NULL;

The comments are left orphaned with task_id = NULL. Useful if the domain accepts "comments with no task" in some flow.

(3) Keep CASCADE but document and test it: if the domain accepts that a hard delete of the parent removes everything (example: a compliance cleanup that genuinely should take the children with it), that's fine. But then the deferred hard delete job has to be documented and tested.

The test that catches the bug

async def test_cascade_no_borra_comments_soft_deleted(session: AsyncSession):
    """
    A hard delete of tasks must NOT break comments with their own soft delete.
    """
    task = Task(author_id=1, title="T")
    session.add(task)
    await session.flush()

    comment = Comment(task_id=task.id, body="Active comment")
    session.add(comment)
    await session.commit()

    # Soft-delete the comment
    comment.soft_delete()
    await session.commit()

    # Hard delete the task (simulating the compliance job)
    await session.execute(delete(Task).where(Task.id == task.id))
    await session.commit()

    # If the FK is RESTRICT, the delete fails. If it's CASCADE, the comment goes.
    # To detect: count comments with the escape hatch
    result = await session.execute(
        select(Comment)
        .where(Comment.id == comment.id)
        .execution_options(include_deleted=True)
    )
    found = result.scalar_one_or_none()

    # The assertion depends on the desired policy:
    # If RESTRICT: the task delete MUST have failed, this test never gets here.
    # If SET NULL: the comment exists with task_id = None.
    # If CASCADE: the comment does NOT exist (lost).
    assert found is not None, "Comment lost to the silent cascade"
    assert found.task_id is None, "task_id should have been set to NULL"

Anti-pattern #3: the JOIN that over-filters

Symptom

Your query lists tasks with their author. You have soft delete on both tables. The listener injects the filter automatically on both sides of the JOIN:

result = await session.execute(
    select(Task, User)
    .join(User, Task.author_id == User.id)
)
# Actual SQL:
# SELECT * FROM tasks JOIN users ON tasks.author_id = users.id
# WHERE tasks.deleted_at IS NULL AND users.deleted_at IS NULL

A customer looks at their tasks. A task they created when they were on a team stopped showing up because the original author (a former employee) was soft-deleted. The task is still active, the author is deleted, and the INNER JOIN removes it.

Why it happens

An INNER JOIN requires a match on both sides. The users.deleted_at IS NULL filter excludes tasks whose author no longer exists (in terms of the automatic filter). It's semantically correct if the product rule is "don't show tasks from deleted users," but that's usually NOT the intent.

How to tell

Any JOIN between two tables with soft delete where the relationship is "parent-child" and the customer expects to see the child even though the parent is no longer visible:

• Tasks created by deleted users.
• Comments on tasks from archived projects.
• Reports that cross events with their actors.

How to fix it

Option A: switch the INNER JOIN to a LEFT JOIN.

result = await session.execute(
    select(Task, User)
    .outerjoin(User, Task.author_id == User.id)
)

But this brings another problem: the listener still filters users.deleted_at IS NULL on the LEFT JOIN, which makes User be NULL when the author is deleted. Better than losing the task, but the query is semantically confusing.

Option B: use the escape hatch for the JOIN.

result = await session.execute(
    select(Task, User)
    .outerjoin(User, Task.author_id == User.id)
    .execution_options(include_deleted=True)  # includes deleted users
    .where(Task.deleted_at.is_(None))  # but filters deleted tasks explicitly
)

Here the escape hatch turns off the automatic filter, and we restore the desired filter for Task explicitly. This is the "exit the default and manually configure what you actually want" pattern.

Option C (the clean pattern): split the queries.

# Query 1: tasks with their author_ids
result = await session.execute(
    select(Task).where(Task.author_id.in_([1, 2, 3]))
)
tasks = result.scalars().all()

# Query 2: users (including deleted ones)
author_ids = {t.author_id for t in tasks}
result = await session.execute(
    select(User)
    .where(User.id.in_(author_ids))
    .execution_options(include_deleted=True)
)
users_by_id = {u.id: u for u in result.scalars().all()}

# Stitching in Python
return [(t, users_by_id.get(t.author_id)) for t in tasks]

More verbose but semantically clear. The listener filters tasks (default), doesn't filter users (escape hatch).

The test that catches the bug

async def test_tasks_de_users_borrados_siguen_visibles(session: AsyncSession):
    user = User(name="Author")
    session.add(user)
    await session.flush()

    task = Task(author_id=user.id, title="Important task")
    session.add(task)
    await session.commit()

    # Soft-delete the user
    user.soft_delete()
    await session.commit()

    # The task has to stay visible
    result = await session.execute(select(Task).where(Task.id == task.id))
    found = result.scalar_one_or_none()
    assert found is not None, "Task lost to the automatic filter in the JOIN"

Anti-pattern #4: raw SQL that escapes the listener

Symptom

Your app has a complex query with CTEs, window functions, and complex joins. The team decides to write it with text():

result = await session.execute(
    text("""
        WITH ranked AS (
            SELECT
                id, title, author_id,
                ROW_NUMBER() OVER (PARTITION BY author_id ORDER BY created_at DESC) AS rn
            FROM tasks
        )
        SELECT id, title FROM ranked WHERE rn <= 5
    """)
)

The listener doesn't fire with text(). The query returns the 5 most recent tasks per author, including deleted ones.

Why it happens

text() runs raw SQL through engine.execute(), not through SQLAlchemy's ORM system. The do_orm_execute listener only intercepts ORM queries (select(), update(), delete() with ORM entities).

How to tell

grep -r "text(" app/ lists every place the team used raw SQL. Each result is a candidate for auditing:

• Does the query touch tables with soft delete?
• Does it have an explicit WHERE deleted_at IS NULL?
• If it's an aggregation or report, can the result be confused with "live" numbers?

How to fix it

Option A: translate it to the ORM if it's viable.

SQLAlchemy 2.0 supports CTEs, window functions, and almost everything you'd do in raw SQL:

from sqlalchemy import func, over

ranked_subquery = (
    select(
        Task.id, Task.title, Task.author_id,
        func.row_number()
            .over(partition_by=Task.author_id, order_by=Task.created_at.desc())
            .label("rn")
    )
    .subquery()
)

result = await session.execute(
    select(ranked_subquery.c.id, ranked_subquery.c.title)
    .where(ranked_subquery.c.rn <= 5)
)

Now the listener injects the filter into the subquery's select(Task...). More verbose but semantically safe.

Option B: if raw SQL is necessary, add the filter manually.

result = await session.execute(
    text("""
        WITH ranked AS (
            SELECT
                id, title, author_id,
                ROW_NUMBER() OVER (PARTITION BY author_id ORDER BY created_at DESC) AS rn
            FROM tasks
            WHERE deleted_at IS NULL  -- ADDED
        )
        SELECT id, title FROM ranked WHERE rn <= 5
    """)
)

Option C: explicitly document that the query does NOT filter (when that's by design).

# Audit query: intentionally includes soft-deleted rows.
# To exclude them, add `WHERE deleted_at IS NULL` to the CTE.
result = await session.execute(text("..."))

The test that catches the bug

async def test_query_compleja_excluye_soft_deleted(session: AsyncSession):
    """
    The top-5-tasks-by-author CTE query has to exclude deleted rows.
    """
    user = User(name="A")
    session.add(user)
    await session.flush()

    # 6 tasks: 5 active + 1 deleted
    tasks = [
        Task(author_id=user.id, title=f"T{i}") for i in range(6)
    ]
    session.add_all(tasks)
    await session.commit()

    tasks[5].soft_delete()
    await session.commit()

    result = await session.execute(top_5_per_author_query())
    # The test has to verify the deleted task is NOT included
    titles = [r.title for r in result]
    assert "T5" not in titles or len(titles) <= 5

Anti-pattern #5: abuse of the include_deleted=True escape hatch

Symptom

Any endpoint with complex queries becomes "easier" if the dev uses execution_options(include_deleted=True) and filters afterward in Python code. What starts as "I need it for auditing" generalizes into "I always use it because it gives me more flexibility." Eventually, an end-customer endpoint uses the escape hatch and returns deleted rows.

Why it happens

The escape hatch exists for legitimate cases (audit, recovery, analytics). But it's trivial to copy-paste. With no enforcement of "only in admin endpoints," it leaks into endpoints that shouldn't use it.

How to tell

grep -rn "include_deleted" app/

List every place. Each one has to have:

1. A comment justifying why it needs to see deleted rows.
2. To be in an admin/audit/recovery module, not in end-customer endpoints.
3. A test that verifies the expected behavior.

How to fix it

Team convention: include_deleted=True only in:

• /admin/* or /internal/* endpoints.
• Audit/recovery background jobs.
• Soft delete-specific tests.

Better: a lint rule or an architecture test.

# tests/architecture/test_no_include_deleted_in_public.py
import re
from pathlib import Path


def test_no_include_deleted_in_public_endpoints():
    """
    `include_deleted=True` is only allowed in admin/internal modules.
    """
    public_dirs = ["app/api/public/", "app/api/v1/"]
    for dir_path in public_dirs:
        for py_file in Path(dir_path).rglob("*.py"):
            content = py_file.read_text()
            if "include_deleted" in content:
                # Allowed if it has a justifying comment
                if not re.search(r"#.*include_deleted.*needed", content):
                    raise AssertionError(
                        f"{py_file} uses include_deleted with no justification. "
                        f"Move it to app/api/admin/ or document it."
                    )

Better still: a specific helper for the legitimate cases.

Instead of exposing the escape hatch directly, offer helpers with semantic names:

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

ModelT = TypeVar("ModelT")


def select_for_audit(model: type[ModelT]) -> Select[tuple[ModelT]]:
    """
    A SELECT that includes soft-deleted rows.
    Use ONLY in audit endpoints or recovery jobs.
    """
    return select(model).execution_options(include_deleted=True)

Now grep -r "select_for_audit" app/ lists the legitimate uses. The helper's name documents the use case. If someone uses the escape hatch directly instead of the helper, code review catches it.

The test that catches the bug

async def test_endpoint_publico_no_devuelve_borradas(client, session):
    """
    GET /tasks (public endpoint) NEVER returns deleted tasks,
    even if the implementer used the escape hatch by mistake.
    """
    t1 = Task(author_id=1, title="Visible")
    t2 = Task(author_id=1, title="HiddenDeleted")
    session.add_all([t1, t2])
    await session.commit()

    t2.soft_delete()
    await session.commit()

    res = await client.get("/tasks")
    titles = [t["title"] for t in res.json()]
    assert "HiddenDeleted" not in titles

Anti-pattern #6: ORDER BY on deleted_at that confuses the plan

Symptom

An audit endpoint: "list the last 100 deleted tasks." The query:

result = await session.execute(
    select(Task)
    .where(Task.deleted_at.is_not(None))
    .order_by(Task.deleted_at.desc())
    .limit(100)
    .execution_options(include_deleted=True)
)

The query is slow. EXPLAIN ANALYZE shows a Seq Scan + Sort.

Why it happens

The partial index is (author_id, created_at DESC) WHERE deleted_at IS NULL. It does NOT cover the "order by deleted_at" case over deleted rows. PostgreSQL has no index for that query; it does a full Seq Scan + sort.

How to tell

EXPLAIN ANALYZE with a Seq Scan on a big table + a Sort in the operation. The giveaway: no existing index covers the query's WHERE + ORDER BY.

How to fix it

Create a partial index on the opposite case (deleted rows):

CREATE INDEX idx_tasks_deleted_audit
  ON tasks (deleted_at DESC)
  WHERE deleted_at IS NOT NULL;

This index is very compact (it only covers 60% of the rows, but that's the fraction it does index) and specific to audit queries. The query improves from 200ms to 1ms.

Trade-off: this index gets updated every time a row goes from active to deleted (the soft delete UPDATE). If deletions are rare, the cost is low. If they're frequent, evaluate whether the frequency of audit queries justifies the overhead.

The test that catches the bug

async def test_audit_query_usa_indice(session: AsyncSession):
    """Verifies the audit query doesn't do a Seq Scan."""
    explain_result = await session.execute(
        text("""
            EXPLAIN
            SELECT * FROM tasks
            WHERE deleted_at IS NOT NULL
            ORDER BY deleted_at DESC LIMIT 100
        """)
    )
    plan = "\n".join(row[0] for row in explain_result)
    assert "Seq Scan" not in plan, f"The audit query does a Seq Scan:\n{plan}"
    assert "idx_tasks_deleted_audit" in plan, f"It doesn't use the audit index:\n{plan}"

Anti-pattern #7: bulk operations that don't respect soft delete

Symptom

A maintenance script runs:

await session.execute(
    update(Task)
    .where(Task.author_id == 42)
    .values(category="legacy")
)

The UPDATE affects both active and deleted tasks. Tasks that were deleted 6 months ago now have category = 'legacy'. If a user later recovers them, they see a category they never had.

Why it happens

The listener doesn't intercept UPDATEs (by design, to allow recovery). But that means any bulk UPDATE affects active AND deleted rows unless you filter explicitly.

How to tell

grep -rn "update(" app/ scripts/ jobs/ and review every bulk UPDATE. Does it filter by deleted_at IS NULL when it should?

How to fix it

A simple rule: any bulk UPDATE on a table with soft delete has to include Model.deleted_at.is_(None) in the WHERE, unless you're explicitly updating deleted rows (recovery, mass un-soft-delete).

await session.execute(
    update(Task)
    .where(Task.author_id == 42, Task.deleted_at.is_(None))
    .values(category="legacy")
)

The test that catches the bug

async def test_update_bulk_no_modifica_borradas(session: AsyncSession):
    t1 = Task(author_id=42, title="Active", category="old")
    t2 = Task(author_id=42, title="Deleted", category="old")
    session.add_all([t1, t2])
    await session.commit()
    t2.soft_delete()
    await session.commit()

    # Bulk update with the defensive filter
    await session.execute(
        update(Task)
        .where(Task.author_id == 42, Task.deleted_at.is_(None))
        .values(category="new")
    )
    await session.commit()

    # Verify
    await session.refresh(t1)
    await session.refresh(t2)
    assert t1.category == "new"
    assert t2.category == "old"  # NOT touched

Why does this capsule matter in real work?

1. The listener protects you from 90% of the bugs, this capsule from the remaining 10%. The anti-patterns you saw aren't prevented by automation: they require knowledge. Without this capsule, your codebase is protected against the forgotten filter in simple queries but vulnerable to the compound cases.

2. It's the capsule that turns you into the team's reviewer. When someone introduces soft delete on a new table, the reviewer checks: (1) the partial index, (2) FKs with cascade audited, (3) stats queries with the filter, (4) JOINs documented. That mental list is what separates a dev who knows from one who applies patterns.

3. Bugs in this category are invisible in QA. Forgotten filters in stats, the silent cascade, JOINs that over-filter — they all pass tests with small data and show up in production. Knowing how to hunt for them in code review is the only defense.

4. It's the material used in senior interviews. "Tell me about a soft delete bug you've debugged" is a common question. This capsule gives you 7 stories ready to tell.

Traps and common mistakes (meta)

Mistake 1 (conceptual): assuming the listener covers everything

Symptom: the team believes that with the listener installed, soft delete is "solved."

Why it's wrong: the listener covers ORM SELECTs. It doesn't cover raw SQL, it doesn't cover UPDATEs/DELETEs, it doesn't solve the silent cascade, it doesn't detect JOINs that over-filter, and it doesn't prevent abuse of the escape hatch. It's necessary but not sufficient.

How to fix it: the golden rule is "automate what you can + train the team on what you can't." This capsule is the training part.

Mistake 2 (practical): code reviews that don't hunt for these patterns

Symptom: PRs with stats queries, joins, or raw SQL get approved without auditing the behavior with soft delete.

Why it happens: the anti-patterns are subtle and don't show up in automated tests. Without a mental checklist, the reviewer lets them through.

How to fix it: a PR checklist when it applies:

• If the query touches a table with soft delete, does it filter correctly?
• If it's a JOIN, is the automatic filter on both tables the desired semantics?
• If it uses text() or raw SQL, does it include WHERE deleted_at IS NULL manually?
• If it uses include_deleted=True, is it justified and in an admin module?
• If it introduces a new FK to a table with soft delete, has the CASCADE/RESTRICT/SET NULL policy been thought through?

Exercises

Exercise 1: audit a codebase

You're given access to an unfamiliar repo with soft delete implemented. List the grep/find commands you'd run to detect the 7 anti-patterns from this capsule. For each one, say what you're looking for.

# 1. Forgotten filter in raw SQL
grep -rn "text(" app/ scripts/ jobs/ | grep -v "test"
# For each result: does it include `deleted_at IS NULL` when it should?

# 2. Silent cascade
psql -c "
SELECT
  tc.table_name AS child_table,
  ccu.table_name AS parent_table,
  rc.delete_rule
FROM information_schema.referential_constraints rc
JOIN information_schema.table_constraints tc ON rc.constraint_name = tc.constraint_name
JOIN information_schema.constraint_column_usage ccu ON rc.unique_constraint_name = ccu.constraint_name
WHERE rc.delete_rule = 'CASCADE'
  AND ccu.table_name IN (SELECT table_name FROM information_schema.columns WHERE column_name = 'deleted_at');
"

# 3. JOINs between tables with soft delete
grep -rn "\.join(" app/ | grep -E "(User|Task|Comment|Project)"
# Review each one: are the filtering semantics on both sides the ones you want?

# 4. Raw SQL in general
grep -rn "execute(text(" app/
# Each one: review it.

# 5. Abuse of the escape hatch
grep -rn "include_deleted" app/
# Each one has to be in an admin/internal module and have a justifying comment.

# 6. ORDER BY on deleted_at with no index
grep -rn "deleted_at.desc\|deleted_at.asc\|order_by(.*deleted" app/
# For each one: is there a partial index on deleted_at IS NOT NULL?

# 7. Bulk UPDATEs with no defensive filter
grep -rn "update(" app/ scripts/ jobs/ | grep -v "test"
# Each one operating on a table with soft delete: does it filter deleted_at IS NULL?

Exercise 2: refactor a stats query

You have this endpoint:

@app.get("/admin/dashboard/totals")
async def dashboard_totals(db: AsyncSession = Depends(get_session)):
    result = await db.execute(text("""
        SELECT
          (SELECT COUNT(*) FROM tasks) AS total_tasks,
          (SELECT COUNT(*) FROM users) AS total_users,
          (SELECT COUNT(*) FROM comments) AS total_comments
    """))
    row = result.first()
    return {
        "tasks": row.total_tasks,
        "users": row.total_users,
        "comments": row.total_comments,
    }

All three tables have soft delete. What's the problem? Refactor it with two different versions (one with the text() corrected, another with the ORM).

@app.get("/admin/dashboard/totals")
async def dashboard_totals(db: AsyncSession = Depends(get_session)):
    result = await db.execute(text("""
        SELECT
          (SELECT COUNT(*) FROM tasks WHERE deleted_at IS NULL) AS total_tasks,
          (SELECT COUNT(*) FROM users WHERE deleted_at IS NULL) AS total_users,
          (SELECT COUNT(*) FROM comments WHERE deleted_at IS NULL) AS total_comments
    """))
    row = result.first()
    return {
        "tasks": row.total_tasks,
        "users": row.total_users,
        "comments": row.total_comments,
    }

from sqlalchemy import func, select

@app.get("/admin/dashboard/totals")
async def dashboard_totals(db: AsyncSession = Depends(get_session)):
    # Each count() goes through the listener: the automatic filter is applied.
    tasks_count = await db.scalar(select(func.count(Task.id)))
    users_count = await db.scalar(select(func.count(User.id)))
    comments_count = await db.scalar(select(func.count(Comment.id)))
    return {
        "tasks": tasks_count,
        "users": users_count,
        "comments": comments_count,
    }

Exercise 3: spot the silent cascade in a given schema

You're shown this schema. List the risks:

CREATE TABLE projects (
  id BIGSERIAL PRIMARY KEY,
  name TEXT NOT NULL,
  deleted_at TIMESTAMPTZ NULL
);

CREATE TABLE tasks (
  id BIGSERIAL PRIMARY KEY,
  project_id BIGINT NOT NULL REFERENCES projects(id) ON DELETE CASCADE,
  title TEXT NOT NULL,
  deleted_at TIMESTAMPTZ NULL
);

CREATE TABLE comments (
  id BIGSERIAL PRIMARY KEY,
  task_id BIGINT NOT NULL REFERENCES tasks(id) ON DELETE CASCADE,
  body TEXT NOT NULL,
  deleted_at TIMESTAMPTZ NULL
);

CREATE TABLE attachments (
  id BIGSERIAL PRIMARY KEY,
  task_id BIGINT NOT NULL REFERENCES tasks(id) ON DELETE SET NULL,
  url TEXT NOT NULL
);

What would you audit and what changes would you propose?

-- For projects → tasks: switch to RESTRICT
ALTER TABLE tasks
  DROP CONSTRAINT tasks_project_id_fkey,
  ADD CONSTRAINT tasks_project_id_fkey
    FOREIGN KEY (project_id) REFERENCES projects(id) ON DELETE RESTRICT;

-- For tasks → comments: switch to RESTRICT
ALTER TABLE comments
  DROP CONSTRAINT comments_task_id_fkey,
  ADD CONSTRAINT comments_task_id_fkey
    FOREIGN KEY (task_id) REFERENCES tasks(id) ON DELETE RESTRICT;

-- For tasks → attachments: fix the NOT NULL + SET NULL conflict
-- Option 1: make the column nullable and keep SET NULL
ALTER TABLE attachments ALTER COLUMN task_id DROP NOT NULL;
-- Option 2: switch to RESTRICT and keep NOT NULL
ALTER TABLE attachments
  DROP CONSTRAINT attachments_task_id_fkey,
  ADD CONSTRAINT attachments_task_id_fkey
    FOREIGN KEY (task_id) REFERENCES tasks(id) ON DELETE RESTRICT;

Exercise 4: write an architecture test

Implement a pytest test that fails if someone uses include_deleted=True in app/api/public/ endpoints.

# tests/architecture/test_no_include_deleted_public.py
import re
from pathlib import Path

import pytest


PUBLIC_DIRS = [
    Path("app/api/public"),
    Path("app/api/v1"),
]


def test_no_include_deleted_in_public_endpoints():
    """
    `include_deleted=True` must not be used in public endpoints.
    Legitimate cases go in app/api/admin/ with a justification.
    """
    violations = []
    for dir_path in PUBLIC_DIRS:
        if not dir_path.exists():
            continue
        for py_file in dir_path.rglob("*.py"):
            content = py_file.read_text()
            for line_num, line in enumerate(content.splitlines(), 1):
                if "include_deleted" not in line:
                    continue
                # Allow it if it's in an explanatory comment
                if line.strip().startswith("#"):
                    continue
                # Allow it if the previous line has a justification
                lines = content.splitlines()
                prev_line = lines[line_num - 2] if line_num >= 2 else ""
                if "# AUDIT-OK:" in prev_line:
                    continue
                violations.append(f"{py_file}:{line_num}: {line.strip()}")

    assert not violations, (
        "include_deleted=True in a public endpoint with no justification.\n"
        "Move it to app/api/admin/ or add a `# AUDIT-OK: <reason>` comment above.\n"
        "Violations:\n" + "\n".join(violations)
    )

# app/api/public/tasks.py
@app.get("/tasks")
async def list_tasks(db: AsyncSession = Depends(get_session)):
    # ...
    result = await db.execute(select(Task))  # OK: automatic filter
    return result.scalars().all()


# app/api/admin/audit.py
@app.get("/admin/audit/all-tasks")
async def all_tasks(db: AsyncSession = Depends(get_session)):
    # AUDIT-OK: admin endpoint for auditing, returns deleted rows intentionally
    result = await db.execute(
        select(Task).execution_options(include_deleted=True)
    )
    return result.scalars().all()

Exercise 5: integrated case — find 3 anti-patterns in some code

Find the anti-patterns in this endpoint:

@app.get("/projects/{project_id}/dashboard")
async def project_dashboard(
    project_id: int,
    db: AsyncSession = Depends(get_session),
):
    # Total tasks in the project
    total = await db.scalar(text(
        f"SELECT COUNT(*) FROM tasks WHERE project_id = {project_id}"
    ))

    # Recent tasks with their author
    result = await db.execute(
        select(Task, User)
        .join(User, Task.author_id == User.id)
        .where(Task.project_id == project_id)
        .order_by(Task.created_at.desc())
        .limit(10)
    )
    recent = result.all()

    # If there are no tasks, return last year's tasks to show some history
    if not recent:
        result = await db.execute(
            select(Task)
            .where(Task.project_id == project_id)
            .execution_options(include_deleted=True)
        )
        recent = [(t, None) for t in result.scalars().all()]

    return {
        "total_tasks": total,
        "recent_tasks": [
            {"id": t.id, "title": t.title, "author": u.name if u else "Unknown"}
            for t, u in recent
        ],
    }

total = await db.scalar(text(
    f"SELECT COUNT(*) FROM tasks WHERE project_id = {project_id}"
))

total = await db.scalar(
    select(func.count(Task.id)).where(Task.project_id == project_id)
)
# The listener injects the soft delete filter automatically.
# And the ORM uses parameterized queries (no SQL injection).

.join(User, Task.author_id == User.id)

result = await db.execute(
    select(Task, User)
    .outerjoin(User, Task.author_id == User.id)
    .execution_options(include_deleted=True)  # include deleted users in the join
    .where(Task.project_id == project_id, Task.deleted_at.is_(None))  # filter tasks explicitly
    .order_by(Task.created_at.desc())
    .limit(10)
)

# Tasks (with the automatic filter)
tasks_result = await db.execute(
    select(Task)
    .where(Task.project_id == project_id)
    .order_by(Task.created_at.desc())
    .limit(10)
)
tasks = tasks_result.scalars().all()

# Users (including deleted ones, escape hatch)
author_ids = {t.author_id for t in tasks}
users_result = await db.execute(
    select(User)
    .where(User.id.in_(author_ids))
    .execution_options(include_deleted=True)
)
users_by_id = {u.id: u for u in users_result.scalars().all()}

recent = [(t, users_by_id.get(t.author_id)) for t in tasks]

if not recent:
    result = await db.execute(
        select(Task)
        .where(Task.project_id == project_id)
        .execution_options(include_deleted=True)  # ← returns deleted rows to the client
    )

return {
    "total_tasks": total,
    "recent_tasks": [
        {"id": t.id, "title": t.title, "author": u.name if u else "Unknown"}
        for t, u in recent
    ],
}

Summary and next step

In this capsule you learned:

• The listener covers 90% of the cases, the anti-patterns cover the remaining 10%. Without this capsule, your codebase is only partially protected.
• 7 concrete anti-patterns with symptom, cause, fix, and the test that detects them:
  1. The forgotten filter in stats queries with text().
  2. The silent cascade in foreign keys.
  3. The INNER JOIN that loses legitimate rows.
  4. Raw SQL that escapes the listener.
  5. Abuse of the include_deleted=True escape hatch.
  6. ORDER BY on deleted_at with no complementary partial index.
  7. Bulk operations that don't filter soft delete.
• Architecture tests and lint rules can prevent convention drift (example: include_deleted only in admin modules).
• A code review checklist specific to soft delete: 5 questions covering the most frequent anti-patterns.

Before moving on you should be able to:

• Audit an unfamiliar codebase with soft delete using the grep/SQL commands from exercise 1.
• Refactor stats queries (text() with a forgotten filter) into versions that filter correctly.
• Audit foreign keys with CASCADE and propose changes when it applies.
• Detect abuse of the escape hatch and propose a mitigation (a specific helper, a lint rule).

Next capsule — Alternatives: archive tables and partitioning. Even though we automated and hardened the pattern, there's a point where soft delete stops scaling: a table with >100M rows, a delete ratio >70%, autovacuum struggling. Capsule 07 teaches you when to migrate to an archive table (moving deleted records to archive.tasks with a nightly job) or partitioning by state (PostgreSQL 11+ partition by list). It's the bridge to guide #14 (Advanced PostgreSQL for Backend), which goes deeper into declarative partitioning.

Resources

1. PostgreSQL Documentation — Foreign Keys — the reference for the CASCADE, RESTRICT, SET NULL, and NO ACTION policies.
2. SQLAlchemy 2.0 — text() and parameter passing — the official reference. The section on security and parameterization is required reading.
3. SQLAlchemy 2.0 — Window functions and CTEs — for translating raw SQL to the ORM when it applies.
4. Cultured Systems — "Avoiding the soft delete anti-pattern" — the critical manifesto. It reinforces why the anti-patterns are so common.
5. Brandur Leach — "Soft deletion probably isn't worth it" — a real case from Stripe about when soft delete stops being sustainable (also relevant to capsule 07).
6. PostgreSQL Wiki — "Don't Do This" — an anthology of anti-patterns in PostgreSQL generally; several apply to soft delete.
7. Crunchy Data — "PostgreSQL constraints: A deeper look" — an analysis of FK constraints and their operational implications.

Module 2 — SQL Patterns for Production APIs Guide

Next capsule: Alternatives — archive tables and partitioning.