Full Project — TaskFlow API

TaskFlow is a complete task management API that uses every concept covered. This is the project you've been building in your head — all the pieces assembled into a real, deployable backend.

Architecture:
• FastAPI app with APIRouter — routes split into auth.py and tasks.py
• SQLModel — User and Task tables with a foreign key relationship
• JWT Authentication — signup, login, protected endpoints
• Full CRUD for tasks — scoped to the authenticated user (users only see their own tasks)
• Pydantic models — separate Create/Read/Update models for each resource
• Dependency chain — get_db → get_current_user → route
• CORS — configured for the Next.js frontend

The User → Task relationship:
• A User has many Tasks (one-to-many)
• Task has owner_id (FK → user.id)
• All task queries are filtered by owner_id = current_user.id
• Users cannot see or modify other users' tasks

Project structure (what a real FastAPI project looks like):
taskflow/
├── main.py          ← app setup, lifespan, include routers, CORS
├── database.py      ← engine, get_db
├── models.py        ← User, Task SQLModel classes
├── schemas.py       ← Pydantic Create/Read/Update models
├── auth.py          ← password hashing, JWT, get_current_user
├── routers/
│   ├── auth.py      ← /auth/signup, /auth/token
│   └── tasks.py     ← /tasks/ CRUD (protected)
└── tests/
    ├── conftest.py
    └── test_tasks.py

This is what you'd actually deploy on a server, connect to a PostgreSQL database, and point your Next.js frontend at. The entire TaskFlow project (FastAPI + SQLModel + Next.js) is the pattern for your personal project.

1# ── models.py ─────────────────────────────────────────
2from sqlmodel import Field, Relationship, SQLModel
3from datetime import datetime, timezone
4
5class User(SQLModel, table=True):
6    id: int | None = Field(default=None, primary_key=True)
7    username: str = Field(unique=True, index=True)
8    email: str = Field(unique=True)
9    hashed_password: str
10    tasks: list["Task"] = Relationship(back_populates="owner")
11
12class Task(SQLModel, table=True):
13    id: int | None = Field(default=None, primary_key=True)
14    title: str
15    description: str | None = None
16    done: bool = False
17    created_at: datetime = Field(default_factory=lambda: datetime.now(timezone.utc))
18    owner_id: int = Field(foreign_key="user.id")
19    owner: User | None = Relationship(back_populates="tasks")
20
21# ── routers/tasks.py ───────────────────────────────────
22from fastapi import APIRouter, Depends, HTTPException
23from sqlmodel import Session, select
24from models import Task, User
25from auth import get_current_user
26from database import get_db
27from typing import Annotated
28
29router = APIRouter(prefix="/tasks", tags=["tasks"])
30DB = Annotated[Session, Depends(get_db)]
31Me = Annotated[User, Depends(get_current_user)]
32
33@router.get("/", response_model=list[Task])
34def list_tasks(db: DB, me: Me, done: bool | None = None):
35    query = select(Task).where(Task.owner_id == me.id)
36    if done is not None:
37        query = query.where(Task.done == done)
38    return db.exec(query).all()
39
40@router.post("/", response_model=Task, status_code=201)
41def create_task(task_in: Task, db: DB, me: Me):
42    task = Task(**task_in.model_dump(exclude={"id", "owner_id"}), owner_id=me.id)
43    db.add(task)
44    db.commit()
45    db.refresh(task)
46    return task
47
48@router.patch("/{task_id}", response_model=Task)
49def update_task(task_id: int, updates: dict, db: DB, me: Me):
50    task = db.get(Task, task_id)
51    if not task or task.owner_id != me.id:
52        raise HTTPException(status_code=404, detail="Task not found")
53    task.sqlmodel_update(updates)
54    db.add(task); db.commit(); db.refresh(task)
55    return task
56
57@router.delete("/{task_id}", status_code=204)
58def delete_task(task_id: int, db: DB, me: Me):
59    task = db.get(Task, task_id)
60    if not task or task.owner_id != me.id:
61        raise HTTPException(status_code=404, detail="Task not found")
62    db.delete(task); db.commit()
63
64# ── main.py ────────────────────────────────────────────
65from fastapi import FastAPI
66from fastapi.middleware.cors import CORSMiddleware
67from contextlib import asynccontextmanager
68from database import create_tables
69from routers import auth, tasks
70
71@asynccontextmanager
72async def lifespan(app: FastAPI):
73    create_tables()   # startup: create DB tables
74    yield
75    pass              # shutdown: cleanup if needed
76
77app = FastAPI(title="TaskFlow", version="1.0", lifespan=lifespan)
78app.add_middleware(CORSMiddleware, allow_origins=["http://localhost:3000"], allow_credentials=True, allow_methods=["*"], allow_headers=["*"])
79app.include_router(auth.router)
80app.include_router(tasks.router)
81
82# Run: uvicorn main:app --reload