Module 8: Final Project — TaskFlow API

Task CRUD + cursor pagination + soft delete

Next →

You've reached the heart of TaskFlow. You'll create the Task model with tenant_id (RLS applied automatically), deleted_at for soft delete, and implement the CRUD endpoints with cursor pagination in GET /tasks. You'll also configure the partial index on deleted_at IS NULL, which is the key optimization for soft delete queries.

By the end, TaskFlow has its central entity operational with the first two patterns applied (cursor + soft delete).

Task model

# app/models/task.py
from datetime import datetime, timezone
import uuid
from typing import Optional

from sqlalchemy import String, Integer, DateTime, ForeignKey, Index
from sqlalchemy.dialects.postgresql import UUID
from sqlalchemy.orm import Mapped, mapped_column, relationship

from app.database import Base


class Task(Base):
    __tablename__ = "tasks"

    id: Mapped[uuid.UUID] = mapped_column(
        UUID(as_uuid=True), primary_key=True, default=uuid.uuid4
    )
    tenant_id: Mapped[uuid.UUID] = mapped_column(
        UUID(as_uuid=True),
        ForeignKey("tenants.id"),
        nullable=False,
    )
    project_id: Mapped[uuid.UUID] = mapped_column(
        UUID(as_uuid=True),
        ForeignKey("projects.id"),
        nullable=False,
    )

    title: Mapped[str] = mapped_column(String(200))
    status: Mapped[str] = mapped_column(String(50), default="pending")

    # Soft delete
    deleted_at: Mapped[Optional[datetime]] = mapped_column(
        DateTime(timezone=True), nullable=True
    )

    # Timestamps
    created_at: Mapped[datetime] = mapped_column(
        DateTime(timezone=True),
        default=lambda: datetime.now(timezone.utc),
    )
    updated_at: Mapped[datetime] = mapped_column(
        DateTime(timezone=True),
        default=lambda: datetime.now(timezone.utc),
        onupdate=lambda: datetime.now(timezone.utc),
    )

    __table_args__ = (
        # Partial index: only live rows, optimizes GET /tasks
        Index(
            "idx_tasks_alive_created",
            "tenant_id",
            "created_at",
            "id",
            postgresql_where="deleted_at IS NULL",
        ),
    )

Add to app/models/__init__.py:

from app.models.task import Task

__all__ = ["Tenant", "User", "Project", "Task"]

Migration with RLS

alembic revision --autogenerate -m "add tasks with soft delete"

Edit it to add RLS:

# alembic/versions/XXX_add_tasks.py
def upgrade() -> None:
    op.create_table(
        'tasks',
        # ... auto-generated ...
    )
    op.create_index(
        'idx_tasks_alive_created',
        'tasks',
        ['tenant_id', 'created_at', 'id'],
        postgresql_where=text('deleted_at IS NULL'),
    )

    # Enable RLS
    op.execute("ALTER TABLE tasks ENABLE ROW LEVEL SECURITY")
    op.execute("""
        CREATE POLICY tenant_isolation_tasks ON tasks
        USING (tenant_id::text = current_setting('app.tenant_id', TRUE))
        WITH CHECK (tenant_id::text = current_setting('app.tenant_id', TRUE))
    """)


def downgrade() -> None:
    op.execute("DROP POLICY tenant_isolation_tasks ON tasks")
    op.execute("ALTER TABLE tasks DISABLE ROW LEVEL SECURITY")
    op.drop_index('idx_tasks_alive_created')
    op.drop_table('tasks')

Basic CRUD endpoints

# app/routers/tasks.py
import uuid
from datetime import datetime, timezone
from typing import Optional

from fastapi import APIRouter, Depends, HTTPException, Query, status, Response, Header
from sqlalchemy.ext.asyncio import AsyncSession
from sqlalchemy import select, and_
from pydantic import BaseModel, Field
import json
from base64 import urlsafe_b64encode, urlsafe_b64decode

from app.deps import get_db_with_tenant, get_current_tenant_id
from app.models import Task


router = APIRouter()


class TaskCreate(BaseModel):
    project_id: uuid.UUID
    title: str = Field(min_length=1, max_length=200)
    status: str = "pending"


class TaskResponse(BaseModel):
    id: uuid.UUID
    project_id: uuid.UUID
    title: str
    status: str
    created_at: datetime
    updated_at: datetime


class TaskListResponse(BaseModel):
    items: list[TaskResponse]
    next_cursor: Optional[str]


def encode_cursor(created_at: datetime, task_id: uuid.UUID) -> str:
    data = {"created_at": created_at.isoformat(), "id": str(task_id)}
    return urlsafe_b64encode(json.dumps(data).encode()).decode()


def decode_cursor(cursor: str) -> dict:
    return json.loads(urlsafe_b64decode(cursor.encode()))


@router.get("/tasks", response_model=TaskListResponse)
async def list_tasks(
    cursor: Optional[str] = None,
    page_size: int = Query(20, ge=1, le=100),
    db: AsyncSession = Depends(get_db_with_tenant),
):
    """Cursor pagination with a composite cursor (created_at DESC, id DESC).

    Filters automatically by tenant_id (RLS) and deleted_at IS NULL (soft delete).
    Uses the partial index idx_tasks_alive_created.
    """
    query = (
        select(Task)
        .where(Task.deleted_at.is_(None))
        .order_by(Task.created_at.desc(), Task.id.desc())
        .limit(page_size + 1)  # +1 to detect more pages
    )

    if cursor:
        try:
            decoded = decode_cursor(cursor)
            cursor_created_at = datetime.fromisoformat(decoded["created_at"])
            cursor_id = uuid.UUID(decoded["id"])
        except (ValueError, KeyError):
            raise HTTPException(400, "Invalid cursor")

        # WHERE (created_at, id) < (cursor_created_at, cursor_id) — ORDER BY DESC
        query = query.where(
            (Task.created_at < cursor_created_at) |
            ((Task.created_at == cursor_created_at) & (Task.id < cursor_id))
        )

    result = await db.execute(query)
    tasks = result.scalars().all()

    has_more = len(tasks) > page_size
    if has_more:
        tasks = tasks[:page_size]

    next_cursor = None
    if has_more and tasks:
        last = tasks[-1]
        next_cursor = encode_cursor(last.created_at, last.id)

    return TaskListResponse(
        items=[TaskResponse(
            id=t.id,
            project_id=t.project_id,
            title=t.title,
            status=t.status,
            created_at=t.created_at,
            updated_at=t.updated_at,
        ) for t in tasks],
        next_cursor=next_cursor,
    )


@router.post("/tasks", response_model=TaskResponse, status_code=status.HTTP_201_CREATED)
async def create_task(
    data: TaskCreate,
    tenant_id: str = Depends(get_current_tenant_id),
    db: AsyncSession = Depends(get_db_with_tenant),
):
    task = Task(
        tenant_id=uuid.UUID(tenant_id),
        project_id=data.project_id,
        title=data.title,
        status=data.status,
    )
    db.add(task)
    await db.commit()
    await db.refresh(task)
    return TaskResponse(
        id=task.id, project_id=task.project_id, title=task.title,
        status=task.status, created_at=task.created_at, updated_at=task.updated_at,
    )


@router.get("/tasks/{task_id}", response_model=TaskResponse)
async def get_task(
    task_id: uuid.UUID,
    db: AsyncSession = Depends(get_db_with_tenant),
):
    task = await db.scalar(
        select(Task).where(
            Task.id == task_id,
            Task.deleted_at.is_(None),  # Don't show soft-deleted
        )
    )
    if not task:
        raise HTTPException(404, "Task not found")

    return TaskResponse(
        id=task.id, project_id=task.project_id, title=task.title,
        status=task.status, created_at=task.created_at, updated_at=task.updated_at,
    )


@router.delete("/tasks/{task_id}", status_code=status.HTTP_204_NO_CONTENT)
async def delete_task(
    task_id: uuid.UUID,
    db: AsyncSession = Depends(get_db_with_tenant),
):
    """Soft delete: sets deleted_at."""
    task = await db.scalar(
        select(Task).where(
            Task.id == task_id,
            Task.deleted_at.is_(None),
        )
    )
    if not task:
        raise HTTPException(404, "Task not found")

    task.deleted_at = datetime.now(timezone.utc)
    await db.commit()

Include it:

# app/main.py
from app.routers import tasks
app.include_router(tasks.router)

Why the partial index matters

Without a partial index:

CREATE INDEX idx_tasks_created ON tasks(tenant_id, created_at, id);

It covers all rows, including the soft-deleted ones. If your table has 30% soft-deleted, the index has 30% of entries you never use.

With a partial index:

CREATE INDEX idx_tasks_alive_created ON tasks(tenant_id, created_at, id)
WHERE deleted_at IS NULL;

It covers only live rows. Smaller, faster to scan.

Check that the planner uses it:

EXPLAIN ANALYZE
SELECT * FROM tasks
WHERE tenant_id = '...' AND deleted_at IS NULL
ORDER BY created_at DESC, id DESC LIMIT 20;
Limit
  ->  Index Scan using idx_tasks_alive_created on tasks
        Index Cond: (tenant_id = '...')

Index Scan using idx_tasks_alive_created confirms it. Without the WHERE deleted_at IS NULL in the query, the planner can't use the partial index.

Anti-pattern: filter in WHERE but not in the INDEX

-- ❌ Index without WHERE
CREATE INDEX idx_tasks_created ON tasks(tenant_id, created_at, id);

-- Query
SELECT * FROM tasks WHERE deleted_at IS NULL ORDER BY created_at DESC LIMIT 20;
-- Plan: Index Scan + Filter (deleted_at IS NULL)
-- Has to scan soft-deleted rows and discard them

vs:

-- ✅ Index with WHERE
CREATE INDEX idx_tasks_alive ON tasks(tenant_id, created_at, id)
WHERE deleted_at IS NULL;

-- Query
SELECT * FROM tasks WHERE deleted_at IS NULL ORDER BY created_at DESC LIMIT 20;
-- Plan: Index Scan (no filter — every row in the index already qualifies)

Tests

# tests/test_tasks_crud.py
import pytest
from httpx import AsyncClient


@pytest.mark.asyncio
async def test_create_and_list_task(client: AsyncClient, tenant_a_token: str, project_a_id: str):
    # Create
    response = await client.post(
        "/tasks",
        json={"project_id": project_a_id, "title": "T1", "status": "pending"},
        headers={"Authorization": f"Bearer {tenant_a_token}"}
    )
    assert response.status_code == 201
    task_id = response.json()["id"]

    # List
    response = await client.get(
        "/tasks",
        headers={"Authorization": f"Bearer {tenant_a_token}"}
    )
    assert response.status_code == 200
    body = response.json()
    assert len(body["items"]) == 1
    assert body["items"][0]["id"] == task_id


@pytest.mark.asyncio
async def test_soft_delete_doesnt_appear_in_list(client, tenant_a_token, project_a_id):
    # Create
    create = await client.post("/tasks", json={...}, headers={...})
    task_id = create.json()["id"]

    # Delete
    delete = await client.delete(f"/tasks/{task_id}", headers={...})
    assert delete.status_code == 204

    # List — must not appear
    list_resp = await client.get("/tasks", headers={...})
    assert len(list_resp.json()["items"]) == 0

    # Direct GET — 404
    get_resp = await client.get(f"/tasks/{task_id}", headers={...})
    assert get_resp.status_code == 404


@pytest.mark.asyncio
async def test_cursor_pagination(client, tenant_a_token, project_a_id):
    # Create 25 tasks
    for i in range(25):
        await client.post(
            "/tasks",
            json={"project_id": project_a_id, "title": f"T{i}", "status": "pending"},
            headers={"Authorization": f"Bearer {tenant_a_token}"}
        )

    # First page
    page1 = await client.get(
        "/tasks?page_size=10",
        headers={"Authorization": f"Bearer {tenant_a_token}"}
    )
    body1 = page1.json()
    assert len(body1["items"]) == 10
    assert body1["next_cursor"] is not None

    # Second page
    cursor = body1["next_cursor"]
    page2 = await client.get(
        f"/tasks?page_size=10&cursor={cursor}",
        headers={"Authorization": f"Bearer {tenant_a_token}"}
    )
    body2 = page2.json()
    assert len(body2["items"]) == 10

    # No overlap between pages
    page1_ids = {t["id"] for t in body1["items"]}
    page2_ids = {t["id"] for t in body2["items"]}
    assert page1_ids.isdisjoint(page2_ids)

    # Third page (final 5)
    page3 = await client.get(
        f"/tasks?page_size=10&cursor={body2['next_cursor']}",
        headers={"Authorization": f"Bearer {tenant_a_token}"}
    )
    body3 = page3.json()
    assert len(body3["items"]) == 5
    assert body3["next_cursor"] is None  # No more pages

Listing with soft-deleted (variant)

Sometimes you need an endpoint that includes soft-deleted rows (admin, audit):

@router.get("/tasks/admin/all", response_model=TaskListResponse)
async def list_all_tasks_including_deleted(
    include_deleted: bool = False,
    cursor: Optional[str] = None,
    page_size: int = Query(20, ge=1, le=100),
    db: AsyncSession = Depends(get_db_with_tenant),
):
    """Admin endpoint that can include deleted tasks."""
    query = (
        select(Task)
        .order_by(Task.created_at.desc(), Task.id.desc())
        .limit(page_size + 1)
    )

    if not include_deleted:
        query = query.where(Task.deleted_at.is_(None))

    # ... rest is the same

Without the deleted_at IS NULL filter, the planner won't use the partial index. For that specific endpoint, consider an additional index without the condition.

Pitfalls and common mistakes

1. Forgetting deleted_at IS NULL in queries.

If a query doesn't filter by deleted_at IS NULL, it returns deleted rows. And worse: it doesn't use the partial index → seq scan on large tables.

2. Hard delete on an already soft-deleted task.

With soft delete active, "delete" is an UPDATE (set deleted_at). If you want a real hard delete (purge), a separate admin endpoint:

@router.delete("/admin/tasks/{task_id}/purge")
async def purge_task(task_id: uuid.UUID, db = Depends(...)):
    # Real hard delete
    await db.execute(delete(Task).where(Task.id == task_id))
    await db.commit()

3. Cursor without a tiebreaker.

If two tasks have the same created_at (rare but possible), a cursor with only created_at can skip one. The id tiebreaker prevents that.

4. Cursor decoded but not validated.

The client can send a cursor with a valid format but arbitrary data. try/except on decode + type validation.

5. UNIQUE constraint on email when there's soft delete.

If a user is deleted and another is created with the same email, it violates the unique constraint.

-- Solution: partial UNIQUE index
CREATE UNIQUE INDEX uq_users_email_alive ON users(email)
WHERE deleted_at IS NULL;

Allows duplicates between soft-deleted and live rows, keeps uniqueness among live rows.

6. __table_args__ without a tuple when there's a single element.

# ❌
__table_args__ = Index(...)

# ✅
__table_args__ = (Index(...),)

7. UPDATE of a soft-deleted row without protection.

PUT /tasks/{id} should return 404 if the task is soft-deleted, not update it. Always filter deleted_at IS NULL in UPDATE queries too.

Summary and next step

What you have now:

  • The Task model with tenant_id, deleted_at, timestamps.
  • A migration with RLS enabled.
  • A partial index on live rows — the key optimization.
  • CRUD endpoints with cursor pagination and soft delete.
  • Tests that verify RLS, cursor, soft delete.

Commit:

git add .
git commit -m "feat: tasks CRUD with cursor pagination and soft delete"

In the next capsule we add the two features that close out the Task model: audit log with PostgreSQL triggers + optimistic locking. You'll create an audit schema with a task_log table, a trigger that captures every change, add a version column with __mapper_args__, and a PUT /tasks/{id} endpoint with an If-Match header → 412 / StaleDataError → 409.

Resources

  1. PostgreSQL — Partial Indexes — official reference.
  2. Markus Winand — Pagination Done the PostgreSQL Way — cursor pagination.
  3. PostgreSQL — UUID type — UUIDs as primary keys.
  4. SQLAlchemy 2.0 — __table_args__ — reference.
  5. Brandur — Soft delete patterns — deep dive.

Capsule 04 of 08 — Module 8 — SQL Patterns for Production APIs Guide

← PreviousMulti-tenancy with Row-Level Security + mock authNext →Audit log with triggers + optimistic locking