Module 2: Doing Soft Deletes Right

Module 2 deliverable: Soft Delete in TaskFlow

What are you going to build and why?

In this project you're going to apply everything you learned in the module to a measurable case: refactor a task management API (mini-TaskFlow) that currently uses hard delete (or soft delete done badly) so it uses the module's complete pattern. You're going to implement the SoftDeleteMixin, the install_soft_delete_filter listener, the partial indexes, and an audit endpoint with an escape hatch. And you're going to measure the before and after with EXPLAIN ANALYZE over a 1M-row table with 60% deleted.

The deliverable is a repository with runnable code, passing tests, and a BENCHMARKS.md that documents the measurements. It's the format expected in any senior backend portfolio: a clear story with numbers backing the decisions.

What you build here gets reused in the capstone project of module 8 (full TaskFlow with multi-tenancy, audit logs, optimistic locking). The module project's code is the "starter kit" for the final project.

Project objective

By the time you complete this project:

You'll have implemented the full soft delete pattern in SQLAlchemy 2.0 + PostgreSQL 16+ + FastAPI 0.110+, portfolio-ready.
You'll have produced a BENCHMARKS.md with real before-and-after measurements (latency, buffers, index size).
You'll have written tests that verify: the automatic filter, the escape hatch, the behavior in JOINs, and that the queries use the partial index.
You'll have documented the design decisions (why mixin + listener, why TIMESTAMPTZ, why the indexes you chose) in a runnable README.

How it fits with what you learned

Module capsule	Where it's applied in the project
02 — Soft delete vs hard delete	The justification in the README: why soft delete is the answer for `tasks` in TaskFlow
03 — Implementing `deleted_at` in PostgreSQL	The schema with the column and the Alembic migration
04 — Mixins and events	The `SoftDeleteMixin` + the listener installed in `app/db.py`
05 — Partial indexes	The partial index in `__table_args__` + the measurement in `BENCHMARKS.md`
06 — Anti-patterns	Tests that verify the filter in stats, the behavior of JOINs, the escape hatch in the admin module
07 — Alternatives	A section in the README: "when this pattern will stop being enough"

Think of the project as the module's "exam." If it passes the tests and produces the right benchmarks, you understood the module.

Technical specifications

Stack

Language: Python 3.11+
Framework: FastAPI 0.110+
ORM: SQLAlchemy 2.0+ with asyncpg
DB: PostgreSQL 16+
Migrations: Alembic 1.13+
Tests: pytest + pytest-asyncio + httpx
Others: Pydantic v2

Initial setup

mkdir taskflow-softdelete
cd taskflow-softdelete

python -m venv venv
source venv/bin/activate

pip install \
  "fastapi[standard]>=0.110" \
  "sqlalchemy[asyncio]>=2.0" \
  "asyncpg>=0.29" \
  "alembic>=1.13" \
  "pydantic>=2.6" \
  "uvicorn[standard]>=0.27" \
  "httpx>=0.27" \
  "pytest>=8.0" \
  "pytest-asyncio>=0.23"

# DB
docker run -d --name pg-taskflow-softdelete \
  -e POSTGRES_DB=taskflow_softdelete \
  -e POSTGRES_PASSWORD=postgres \
  -p 5432:5432 \
  postgres:16

File structure

taskflow-softdelete/
├── app/
│   ├── __init__.py
│   ├── main.py                 # FastAPI app + endpoints
│   ├── db.py                   # engine + session + listener install
│   ├── db_setup/
│   │   ├── __init__.py
│   │   └── soft_delete_filter.py  # install_soft_delete_filter
│   ├── models/
│   │   ├── __init__.py
│   │   ├── base.py
│   │   ├── mixins.py           # SoftDeleteMixin
│   │   └── task.py
│   ├── schemas/
│   │   ├── __init__.py
│   │   └── task.py             # TaskOut, TaskCreate
│   └── api/
│       ├── __init__.py
│       ├── public/
│       │   └── tasks.py        # public endpoints: they use the automatic filter
│       └── admin/
│           └── audit.py        # admin endpoints: they use the escape hatch
├── alembic/
│   ├── env.py
│   └── versions/
│       └── xxx_initial_schema.py
├── tests/
│   ├── conftest.py
│   ├── test_soft_delete_filter.py
│   ├── test_partial_index.py
│   └── test_anti_patterns.py
├── scripts/
│   ├── seed.py                 # populate the table with 1M tasks, 60% deleted
│   └── benchmark.py            # measurements for BENCHMARKS.md
├── alembic.ini
├── pyproject.toml
├── BENCHMARKS.md               # the project's main deliverable
└── README.md

Required functionality

1. `Task` model with `SoftDeleteMixin`

Model spec:
- Table: tasks
- Columns: id (BIGINT PK), author_id (BIGINT NOT NULL), title (VARCHAR(200)),
  created_at (TIMESTAMPTZ NOT NULL DEFAULT now()), deleted_at (TIMESTAMPTZ NULL via the mixin)
- Partial index: (author_id, created_at DESC) WHERE deleted_at IS NULL

Expected behavior:

The mixin has to be in app/models/mixins.py and be reusable for other future models (example: Comment, Project).
The partial index has to be defined in __table_args__ with postgresql_where=text(...).
The soft_delete(), restore() methods and the is_deleted property are available.

2. The `install_soft_delete_filter` listener

Listener spec:
- An install_soft_delete_filter(session_class) function that registers do_orm_execute
- It injects `WHERE deleted_at IS NULL` into SELECTs over SoftDeleteMixin entities
- Escape hatch: execution_options(include_deleted=True) turns it off
- It works with joins (include_aliases=True)
- It does NOT get registered twice (a guard against re-registration)

Expected behavior:

Without the escape hatch, no SELECT query can see soft-deleted rows.
With the escape hatch, all the soft-deleted rows are visible.
UPDATE and DELETE aren't filtered (by design, to allow recovery).

3. Public endpoints in `app/api/public/tasks.py`

Endpoint spec:

GET /tasks
  Query params: author_id (optional), limit (default 50, max 200)
  Response: List[TaskOut] (active only, automatic filter)
  Status: 200

POST /tasks
  Body: TaskCreate (author_id, title)
  Response: TaskOut (created)
  Status: 201

DELETE /tasks/{task_id}
  Soft delete (UPDATE SET deleted_at = NOW())
  Status: 204 if OK, 404 if it doesn't exist, 410 if already deleted

PUT /tasks/{task_id}/restore
  Restore (UPDATE SET deleted_at = NULL)
  Requires include_deleted=True to find the row to restore
  Status: 200 if OK, 404 if it doesn't exist, 422 if it wasn't deleted

Expected behavior:

None of these endpoints uses include_deleted=True except restore (which needs it to find the deleted row before restoring it).
The test_no_include_deleted_in_public test (anti-pattern #5 from capsule 06) has to pass.

4. Admin endpoint in `app/api/admin/audit.py`

Spec:

GET /admin/audit/tasks/all
  Returns every task (active + deleted)
  Uses execution_options(include_deleted=True)
  A comment justifying the use of the escape hatch
  Status: 200

GET /admin/audit/tasks/recently-deleted?days=30
  Returns tasks soft-deleted in the last N days
  Uses execution_options(include_deleted=True)
  Status: 200

Expected behavior:

These endpoints DO use include_deleted=True with a justification in a comment.
They're in app/api/admin/, not in app/api/public/.

5. Alembic migration with `CREATE INDEX CONCURRENTLY`

Spec:
- The initial migration creates the table and the deleted_at column.
- A separate migration creates the partial index with CREATE INDEX CONCURRENTLY.
- A symmetric downgrade.

6. A seed script of 1M tasks (60% deleted)

Spec:
- python scripts/seed.py inserts 1M tasks spread across 1000 author_ids.
- Then it marks 60% as soft-deleted (UPDATE SET deleted_at = ...).
- Output: total time + counts (active vs deleted).

7. Benchmark script

Spec:
- python scripts/benchmark.py runs:
  (a) DROP the partial index; CREATE a normal index; measure the typical query with EXPLAIN ANALYZE.
  (b) DROP the normal index; CREATE the partial index; measure the same query.
  (c) Report: Execution Time, Buffers, Rows Removed by Filter for each case.
- Output: prints a comparison table + writes to BENCHMARKS.md

8. Required tests

test_soft_delete_filter.py: verifies the automatic filter and the escape hatch.
test_partial_index.py: verifies the queries use the partial index (EXPLAIN doesn't show a Seq Scan).
test_anti_patterns.py: verifies that stats don't count deleted rows, that JOINs behave as designed, and that include_deleted doesn't appear in app/api/public/.

Validation and error handling

What has to be validated

POST /tasks: title not empty, author_id positive.
DELETE /tasks/{id}: the id exists, it wasn't already deleted.
PUT /tasks/{id}/restore: the id exists, it was deleted.
GET /tasks?limit=: limit between 1 and 200.
GET /admin/audit/tasks/recently-deleted?days=: days between 1 and 365.

Errors that have to be handled

Task not found (404): when the id doesn't exist.
Task already deleted (410 Gone): when you try to delete one that's already deleted.
Task wasn't deleted (422): when you try to restore an active one.
Validation fails (422): Pydantic errors.
DB connection error (503): if PostgreSQL doesn't respond, return "service unavailable."

Minimal implementation example

This is the skeleton. Your job is to expand it to cover all the functionality.

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


class SoftDeleteMixin:
    @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_setup/soft_delete_filter.py
from sqlalchemy import event
from sqlalchemy.orm import Session, with_loader_criteria

from app.models.mixins import SoftDeleteMixin

_REGISTERED: set[type[Session]] = set()


def install_soft_delete_filter(target_session_class: type[Session]) -> None:
    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) -> 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/main.py
from fastapi import FastAPI
from app.api.public import tasks as public_tasks
from app.api.admin import audit as admin_audit

app = FastAPI(title="TaskFlow Soft Delete")
app.include_router(public_tasks.router, prefix="/tasks", tags=["tasks"])
app.include_router(admin_audit.router, prefix="/admin/audit", tags=["admin"])


@app.get("/health")
def health():
    return {"status": "ok"}

To run it:

# Migrations
alembic upgrade head

# Seed
python scripts/seed.py

# Benchmark
python scripts/benchmark.py

# App
uvicorn app.main:app --reload

# Tests
pytest tests/ -v

Evaluation rubric (self-check)

Functionality (40 points)

(5 pts) SoftDeleteMixin defined and reusable.
(5 pts) The install_soft_delete_filter listener works and is idempotent.
(5 pts) The Task model with the partial index in __table_args__.
(5 pts) The GET /tasks endpoint filters automatically.
(5 pts) The DELETE /tasks/{id} endpoint does a soft delete with "already deleted" handling.
(5 pts) The PUT /tasks/{id}/restore endpoint works with include_deleted=True.
(5 pts) Admin endpoints in app/api/admin/ with a justified escape hatch.
(5 pts) The Alembic migration with CREATE INDEX CONCURRENTLY and autocommit_block.

Tests (20 points)

(5 pts) test_filtro_automatico_excluye_soft_deleted passes.
(5 pts) test_escape_hatch_devuelve_soft_deleted passes.
(5 pts) test_query_usa_indice_parcial (parsing EXPLAIN ANALYZE) passes.
(5 pts) test_no_include_deleted_in_public (the architecture test) passes.

Benchmarks (20 points)

(5 pts) The scripts/benchmark.py script runs with no errors.
(5 pts) BENCHMARKS.md documents the latency with a normal index vs a partial one.
(5 pts) BENCHMARKS.md reports Buffers: shared hit and Rows Removed by Filter.
(5 pts) BENCHMARKS.md includes analysis (not just numbers): why the improvement is the one observed.

Documentation (15 points)

(5 pts) The README documents the setup, the commands, the design decisions.
(5 pts) Comments in the code justify the use of the escape hatch in every admin location.
(5 pts) BENCHMARKS.md has context (hardware, PostgreSQL version, config).

Extra credit (optional, up to +15 pts)

(+5 pts) Also implement the Comment model with SoftDeleteMixin and verify the behavior in a tasks-comments JOIN.
(+5 pts) Create an additional partial index for audit queries (WHERE deleted_at IS NOT NULL) and measure its impact.
(+5 pts) Implement middleware that logs the use of include_deleted=True (an audit of the audit).

Total: 95 points Pass: ≥66 points (70%) Outstanding: ≥80 points

Common mistakes in this project

Mistake 1: the listener doesn't work because it gets registered AFTER the first `select()`

Symptom: the tests fail, the queries return soft-deleted rows even though "the listener is installed."

Why it happens: you call install_soft_delete_filter(AsyncSession) after having created the engine and run queries. The listener gets registered for new sessions, but the ones that already exist aren't affected.

How to fix it: register the listener IMMEDIATELY after creating the async_sessionmaker, before any real session. Ideally in the setup module, not in a lazy init.

Mistake 2: the partial index is NOT created with `Base.metadata.create_all()`

Symptom: you run create_all() in setup but \d tasks doesn't show the partial index.

Why it happens: incorrect __table_args__ syntax. If you defined it as a bare dict or as a loose attribute, SQLAlchemy ignores it.

How to fix it: use the tuple correctly:

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

Mistake 3: the `restore` endpoint can't find the task because the automatic filter hides it

Symptom: PUT /tasks/{id}/restore always returns 404 for soft-deleted tasks.

Why it happens: the query to find the task uses the automatic filter, which excludes it because it's deleted.

How to fix it: use the escape hatch in restore:

result = await session.execute(
    select(Task)
    .where(Task.id == task_id)
    .execution_options(include_deleted=True)  # CRITICAL
)
task = result.scalar_one_or_none()
if task is None:
    raise HTTPException(404, "Task not found")
if not task.is_deleted:
    raise HTTPException(422, "Task is not deleted")
task.restore()
await session.commit()

Mistake 4: the benchmark measures cached data and the numbers aren't representative

Symptom: the first run of the benchmark gives 50ms, the following ones give 0.5ms. Which one do you report?

Why it happens: PostgreSQL caches pages in shared_buffers. The first query is "cold," the following ones are "warm."

How to fix it:

Document whether you're measuring cold or warm (typically warm, because it represents sustained load).
Run the query 5 times before measuring (warmup), then 5 times while measuring, and report the median.
Restart PostgreSQL (docker restart pg-taskflow-softdelete) between setups if you want comparable cold measurements.

Module 1 of guide #12 (Database Performance & Query Tuning) goes deeper into this in its "Reproducible baselines" capsule.

Mistake 5: forgetting the defensive filter in the soft delete UPDATE

Symptom: calling DELETE /tasks/{id} twice also updates the deleted_at the second time, losing the original date.

Why it happens: the query is UPDATE SET deleted_at = NOW() WHERE id = $1 without AND deleted_at IS NULL.

How to fix it:

result = await session.execute(
    update(Task)
    .where(Task.id == task_id, Task.deleted_at.is_(None))
    .values(deleted_at=func.now())
)
if result.rowcount == 0:
    # Either it doesn't exist, or it was already deleted
    raise HTTPException(...)

What to do if you get stuck

Setup fails with asyncpg not installed: make sure you install asyncpg>=0.29 and use postgresql+asyncpg:// in DATABASE_URL.
The migration fails with CREATE INDEX CONCURRENTLY inside a transaction: you're forgetting with op.get_context().autocommit_block():. Review capsule 05.
Tests pass locally but fail in CI: the session class is probably the wrong one. Make sure install_soft_delete_filter gets registered with AsyncSession (the class, not an instance).
The benchmark gives numbers very different from the module's: normal. Hardware, PostgreSQL configuration, and the exact data all vary. What matters is the shape: the partial index should be ~10-100x better than a normal one on a table with a high delete ratio.
I don't know what to put in BENCHMARKS.md: see the template below.

`BENCHMARKS.md` template

# BENCHMARKS — TaskFlow Soft Delete

## Context

- Hardware: MacBook Pro M2, 32GB RAM
- PostgreSQL: 16.2 (Docker, default config)
- Python: 3.11.7
- SQLAlchemy: 2.0.27
- Data: 1,000,000 tasks, 600,000 soft-deleted (60%), 1000 distinct author_ids

## Query measured

```sql
SELECT id, title FROM tasks
WHERE author_id = 42 AND deleted_at IS NULL
ORDER BY created_at DESC LIMIT 50;

Run 5 times after 5 warmup runs. Reported: the median.

Results

Setup	Plan	Execution Time	Buffers (shared hit)	Rows Removed by Filter	Index size
No index	Seq Scan + Sort	285 ms	18,402	N/A	0 (no index)
Normal index `(author_id, created_at)`	Index Scan + Filter	28.3 ms	2,412	1,483	32 MB
Partial index `WHERE deleted_at IS NULL`	Index Scan	0.48 ms	54	0	13 MB

Analysis

The partial index is 59x faster than the normal index in this scenario.
The improvement ratio correlates with the delete ratio: with 60% of rows deleted, the normal index walks ~2.5x more rows than the partial one.
The index size is proportional to the indexed rows: normal index = 32MB (1M rows), partial index = 13MB (400k active rows).
The buffers read drop 45x with the partial index: it confirms the main saving is IO.

Expected production impact

For a table with a similar profile:

The GET /tasks endpoint drops from p95=30-100ms (with a normal index) to p95=1-2ms (with a partial one).
You reduce pressure on shared_buffers: the partial index fits in cache, better cache hit ratio.
You reduce maintenance time (VACUUM, REINDEX) proportionally.

References

Capsule 03 of module 2: Implementing deleted_at in PostgreSQL.
Capsule 05 of module 2: Partial indexes for soft deletes.


---

## Resources for the project

1. [SQLAlchemy 2.0 — `with_loader_criteria`](https://docs.sqlalchemy.org/en/20/orm/queryguide/api.html#sqlalchemy.orm.with_loader_criteria) — the primitive the listener uses.
2. [SQLAlchemy 2.0 — Async Quickstart](https://docs.sqlalchemy.org/en/20/orm/extensions/asyncio.html) — the reference for the async session, engine, dependency injection.
3. [Alembic — `op.execute` and `autocommit_block`](https://alembic.sqlalchemy.org/en/latest/api/runtime.html#alembic.runtime.migration.MigrationContext.autocommit_block) — for the migration with `CREATE INDEX CONCURRENTLY`.
4. [PostgreSQL Documentation — Partial Indexes](https://www.postgresql.org/docs/current/indexes-partial.html) — the official reference.
5. [pytest-asyncio documentation](https://pytest-asyncio.readthedocs.io/) — for async tests.
6. [FastAPI — Dependency injection](https://fastapi.tiangolo.com/tutorial/dependencies/) — for `Depends(get_session)`.

---

## What comes next

What you built here is the foundation of the capstone project of module 8 (full TaskFlow). The same `SoftDeleteMixin` and `install_soft_delete_filter` get reused as-is. What gets added in module 8:

- **Multi-tenancy with RLS** (module 4): `tenant_id` in every table, PostgreSQL policies that isolate data between tenants.
- **An audit log with triggers** (module 3): every soft delete fires an event into `audit.task_history`.
- **Optimistic locking** (module 6): a version column on Task, handling `StaleDataError`.
- **Cursor pagination** (module 1): the `GET /tasks` endpoint uses a cursor instead of an offset.
- **Bulk operations** (module 7): a `POST /tasks/bulk` endpoint with COPY.
- **Zero-downtime migration** (module 5): adding a `priority` column to a table with live traffic.

Before moving on to module 3, make sure your project:

- Passes all the required tests.
- Has a `BENCHMARKS.md` with real numbers (not copy-pasted from the module's).
- Has a README documenting your key technical decision: why you chose mixin + listener (it should say "because it scales to teams with no extra discipline and the escape hatch is greppable").
- Is runnable end-to-end with a single command documented in the README.

If you passed everything: welcome to module 3. It starts with audit logs and history tables, the natural complement to soft delete.

---

*Module 2 — SQL Patterns for Production APIs Guide*

**Next module:** Audit Logs and History Tables.

← PreviousAlternatives: archive tables and partitioning Next →Audit Logs and History Tables — Module introduction

Module 2 deliverable: Soft Delete in TaskFlow

What are you going to build and why?

Project objective

How it fits with what you learned

Technical specifications

Stack

Initial setup

File structure

Required functionality

1. Task model with SoftDeleteMixin

2. The install_soft_delete_filter listener

3. Public endpoints in app/api/public/tasks.py

4. Admin endpoint in app/api/admin/audit.py

5. Alembic migration with CREATE INDEX CONCURRENTLY

6. A seed script of 1M tasks (60% deleted)

7. Benchmark script

8. Required tests

Validation and error handling

What has to be validated

Errors that have to be handled

Minimal implementation example

Evaluation rubric (self-check)

Functionality (40 points)

Tests (20 points)

Benchmarks (20 points)

Documentation (15 points)

Extra credit (optional, up to +15 pts)

Common mistakes in this project

Mistake 1: the listener doesn't work because it gets registered AFTER the first select()

Mistake 2: the partial index is NOT created with Base.metadata.create_all()

Mistake 3: the restore endpoint can't find the task because the automatic filter hides it

Mistake 4: the benchmark measures cached data and the numbers aren't representative

Mistake 5: forgetting the defensive filter in the soft delete UPDATE

What to do if you get stuck

BENCHMARKS.md template

Results

Analysis

Expected production impact

References

1. `Task` model with `SoftDeleteMixin`

2. The `install_soft_delete_filter` listener

3. Public endpoints in `app/api/public/tasks.py`

4. Admin endpoint in `app/api/admin/audit.py`

5. Alembic migration with `CREATE INDEX CONCURRENTLY`

Mistake 1: the listener doesn't work because it gets registered AFTER the first `select()`

Mistake 2: the partial index is NOT created with `Base.metadata.create_all()`

Mistake 3: the `restore` endpoint can't find the task because the automatic filter hides it

`BENCHMARKS.md` template