I am a Data Engineer and AI Engineer who also trains developers. A few weeks ago I was preparing a FastAPI session for a group of students and I wanted a project that was simple enough to understand fully, but rich enough to cover all the important concepts in one place.
The result was this Navas Task Manager API - and by the end of this article, you will have built it too.
We will cover:
- Project structure and why it matters
- Connecting to SQLite with SQLAlchemy
- Defining database models
- Validating data with Pydantic schemas
- Separating database logic into a
crud.pylayer - Building full CRUD endpoints with APIRouter
- Adding custom middleware for request logging
- Configuring CORS so a frontend can talk to the API
- Testing everything live in the browser - no Postman needed
The full project is on GitHub at the end of this article. Let's build it.
What We Are Building
A Task Manager API where users can create an account and manage their tasks. Two resources, full CRUD on both, one relationship between them.
Users → Tasks (one user can have many tasks)
The final API will have 10 endpoints:
| Method | URL | What it does |
|---|---|---|
| POST | /users/ | Create a user |
| GET | /users/ | List all users |
| GET | /users/{id} | Get one user + their tasks |
| PUT | /users/{id} | Update a user |
| DELETE | /users/{id} | Delete a user |
| POST | /tasks/ | Create a task |
| GET | /tasks/ | List all tasks |
| GET | /tasks/{id} | Get one task |
| PUT | /tasks/{id} | Update a task |
| DELETE | /tasks/{id} | Delete a task |
Why FastAPI?
Before we write any code, let me answer the question beginners always ask.
FastAPI is a modern Python web framework built on top of Starlette and Pydantic. Here is what makes it stand out:
Speed - FastAPI is one of the fastest Python frameworks available, on par with Node.js and Go for async workloads.
Automatic docs - visit /docs and FastAPI generates interactive API documentation automatically. You can test every endpoint right there in the browser.
Type safety - FastAPI uses Python type hints to validate incoming data, catch errors early, and generate accurate documentation - all from the same code.
Developer experience - less boilerplate than Django, more structure than Flask, and excellent error messages that tell you exactly what went wrong.
Project Structure
Here is the folder layout we will build:
task_manager/
├── main.py ← app entry point, middleware, CORS, routers
├── database.py ← SQLite connection and session management
├── models.py ← SQLAlchemy table definitions
├── schemas.py ← Pydantic request/response schemas
├── crud.py ← all database operations
├── tasks.db ← auto-created SQLite file on first run
└── routers/
├── __init__.py
├── tasks.py ← task endpoints
└── users.py ← user endpoints
This structure follows the separation of concerns principle -each file has one clear job. When something breaks, you know exactly which file to open.
Installation
mkdir task_manager
cd task_manager
mkdir routers
touch routers/__init__.py
pip install fastapi uvicorn sqlalchemy
Step 1 - database.py
This is the foundation. Everything else sits on top of it.
from sqlalchemy import create_engine
from sqlalchemy.ext.declarative import declarative_base
from sqlalchemy.orm import sessionmaker
DATABASE_URL = "sqlite:///./tasks.db"
engine = create_engine(
DATABASE_URL, connect_args={"check_same_thread": False}
)
SessionLocal = sessionmaker(autocommit=False, autoflush=False, bind=engine)
Base = declarative_base()
def get_db():
db = SessionLocal()
try:
yield db
finally:
db.close()
What each part does:
DATABASE_URL - tells SQLAlchemy where the database lives. The ./ means create the file in the current folder. Switching to PostgreSQL later is as simple as changing this one string.
engine - the actual connection to the database. The check_same_thread: False argument is SQLite-specific and needed because FastAPI handles requests across multiple threads.
SessionLocal - a factory that creates database sessions. Think of a session like a shopping cart for database operations - you collect your changes, then commit them all at once.
Base - the parent class all database models will inherit from. SQLAlchemy uses it to track which Python classes map to which database tables.
get_db() - a dependency function. The yield keyword is the key here. FastAPI opens a session, yields it to the endpoint function, runs the endpoint, then the finally block closes the session automatically - even if an error occurs.
Step 2 - models.py
Models are Python classes that become database tables.
from sqlalchemy import Column, Integer, String, Boolean, ForeignKey
from sqlalchemy.orm import relationship
from database import Base
class User(Base):
__tablename__ = "users"
id = Column(Integer, primary_key=True, index=True)
name = Column(String, nullable=False)
email = Column(String, unique=True, index=True)
tasks = relationship("Task", back_populates="owner")
class Task(Base):
__tablename__ = "tasks"
id = Column(Integer, primary_key=True, index=True)
title = Column(String, nullable=False)
description = Column(String, nullable=True)
completed = Column(Boolean, default=False)
owner_id = Column(Integer, ForeignKey("users.id"))
owner = relationship("User", back_populates="tasks")
The important parts:
class User(Base) - inheriting from Base registers this class with SQLAlchemy. The __tablename__ attribute sets the actual name of the table in the database file.
ForeignKey("users.id") - this is how the one-to-many relationship is enforced at the database level. Every task must belong to a user.
relationship() - this is where the magic happens. Writing user.tasks in Python will automatically fetch all tasks belonging to that user. SQLAlchemy handles the SQL query behind the scenes.
Step 3 : schemas.py
This is where most beginners get confused - "why do I need schemas if I already have models?"
Here is the answer:
| SQLAlchemy Model | Pydantic Schema |
|---|---|
| Talks to the database | Talks to the API user |
| Defines table columns | Validates request data |
| Never shown directly | Shapes the response |
They are separate because what you store in the database and what you expose through the API are often different. You might store a hashed password but never want to return it in a response. Schemas give you that control.
from pydantic import BaseModel
from typing import Optional, List
# ─── TASK SCHEMAS ───────────────────────────────
class TaskBase(BaseModel):
title: str
description: Optional[str] = None
class TaskCreate(TaskBase):
owner_id: int
class TaskUpdate(BaseModel):
title: Optional[str] = None
description: Optional[str] = None
completed: Optional[bool] = None
class TaskResponse(TaskBase):
id: int
completed: bool
owner_id: int
class Config:
from_attributes = True
# ─── USER SCHEMAS ───────────────────────────────
class UserBase(BaseModel):
name: str
email: str
class UserCreate(UserBase):
pass
class UserUpdate(BaseModel):
name: Optional[str] = None
email: Optional[str] = None
class UserResponse(UserBase):
id: int
tasks: List[TaskResponse] = []
class Config:
from_attributes = True
The 3-schema pattern:
-
Base- shared fields used by both Create and Response -
Create- fields the user sends to the API -
Update- all fields areOptionalso users only send what they want to change -
Response- what the API sends back, including database-generated fields likeid
from_attributes = True is the bridge between SQLAlchemy and Pydantic. Without it, Pydantic cannot read a SQLAlchemy model object - it only understands plain dictionaries by default.
Notice UserResponse includes tasks: List[TaskResponse]. When you fetch a user, their tasks come nested inside the response automatically.
Step 4 - crud.py
All database reads and writes live here. Endpoints never touch the database directly.
Think of it this way - crud.py is the kitchen, routers are the waiters. The waiter takes the order (request), passes it to the kitchen (crud), and brings back the plate (response). Waiters don't cook.
from sqlalchemy.orm import Session
import models, schemas
# ─── USERS ──────────────────────────────────────
def create_user(db: Session, user: schemas.UserCreate):
db_user = models.User(name=user.name, email=user.email)
db.add(db_user)
db.commit()
db.refresh(db_user)
return db_user
def get_user(db: Session, user_id: int):
return db.query(models.User).filter(models.User.id == user_id).first()
def get_users(db: Session):
return db.query(models.User).all()
def update_user(db: Session, user_id: int, data: schemas.UserUpdate):
db_user = db.query(models.User).filter(models.User.id == user_id).first()
if not db_user:
return None
updates = data.model_dump(exclude_unset=True)
for field, value in updates.items():
setattr(db_user, field, value)
db.commit()
db.refresh(db_user)
return db_user
def delete_user(db: Session, user_id: int):
db_user = db.query(models.User).filter(models.User.id == user_id).first()
if not db_user:
return None
db.delete(db_user)
db.commit()
return db_user
# ─── TASKS ──────────────────────────────────────
def create_task(db: Session, task: schemas.TaskCreate):
db_task = models.Task(**task.model_dump())
db.add(db_task)
db.commit()
db.refresh(db_task)
return db_task
def get_task(db: Session, task_id: int):
return db.query(models.Task).filter(models.Task.id == task_id).first()
def get_tasks(db: Session):
return db.query(models.Task).all()
def update_task(db: Session, task_id: int, data: schemas.TaskUpdate):
db_task = db.query(models.Task).filter(models.Task.id == task_id).first()
if not db_task:
return None
updates = data.model_dump(exclude_unset=True)
for field, value in updates.items():
setattr(db_task, field, value)
db.commit()
db.refresh(db_task)
return db_task
def delete_task(db: Session, task_id: int):
db_task = db.query(models.Task).filter(models.Task.id == task_id).first()
if not db_task:
return None
db.delete(db_task)
db.commit()
return db_task
The 3-step create pattern - every create operation follows this exactly:
-
db.add(obj)-stages the record (puts it in the cart) -
db.commit()- saves to disk permanently -
db.refresh(obj)- reloads the object so you get the auto-generatedidback
model_dump(exclude_unset=True) on updates is important. It only includes fields the user actually sent in the request. Without it, unset optional fields would come through as None and accidentally overwrite existing data.
Step 5: The Routers
Create two files: routers/tasks.py and routers/users.py.
routers/tasks.py
from fastapi import APIRouter, Depends, HTTPException
from sqlalchemy.orm import Session
from typing import List
import crud, schemas
from database import get_db
router = APIRouter(prefix="/tasks", tags=["Tasks"])
@router.post("/", response_model=schemas.TaskResponse)
def create_task(task: schemas.TaskCreate, db: Session = Depends(get_db)):
return crud.create_task(db, task)
@router.get("/", response_model=List[schemas.TaskResponse])
def get_tasks(db: Session = Depends(get_db)):
return crud.get_tasks(db)
@router.get("/{task_id}", response_model=schemas.TaskResponse)
def get_task(task_id: int, db: Session = Depends(get_db)):
task = crud.get_task(db, task_id)
if not task:
raise HTTPException(status_code=404, detail="Task not found")
return task
@router.put("/{task_id}", response_model=schemas.TaskResponse)
def update_task(task_id: int, data: schemas.TaskUpdate, db: Session = Depends(get_db)):
task = crud.update_task(db, task_id, data)
if not task:
raise HTTPException(status_code=404, detail="Task not found")
return task
@router.delete("/{task_id}")
def delete_task(task_id: int, db: Session = Depends(get_db)):
task = crud.delete_task(db, task_id)
if not task:
raise HTTPException(status_code=404, detail="Task not found")
return {"message": "Task deleted successfully"}
routers/users.py
from fastapi import APIRouter, Depends, HTTPException
from sqlalchemy.orm import Session
from typing import List
import crud, schemas
from database import get_db
router = APIRouter(prefix="/users", tags=["Users"])
@router.post("/", response_model=schemas.UserResponse)
def create_user(user: schemas.UserCreate, db: Session = Depends(get_db)):
return crud.create_user(db, user)
@router.get("/", response_model=List[schemas.UserResponse])
def get_users(db: Session = Depends(get_db)):
return crud.get_users(db)
@router.get("/{user_id}", response_model=schemas.UserResponse)
def get_user(user_id: int, db: Session = Depends(get_db)):
user = crud.get_user(db, user_id)
if not user:
raise HTTPException(status_code=404, detail="User not found")
return user
@router.put("/{user_id}", response_model=schemas.UserResponse)
def update_user(user_id: int, data: schemas.UserUpdate, db: Session = Depends(get_db)):
user = crud.update_user(db, user_id, data)
if not user:
raise HTTPException(status_code=404, detail="User not found")
return user
@router.delete("/{user_id}")
def delete_user(user_id: int, db: Session = Depends(get_db)):
user = crud.delete_user(db, user_id)
if not user:
raise HTTPException(status_code=404, detail="User not found")
return {"message": "User deleted successfully"}
Three concepts to understand in the routers:
APIRouter(prefix="/tasks", tags=["Tasks"]) - a mini-app for one resource. The prefix means every route here automatically starts with /tasks. The tags group them neatly on the /docs page.
db: Session = Depends(get_db) - FastAPI's dependency injection. Instead of manually creating a database session in every function, FastAPI calls get_db(), hands you a fresh session, and closes it automatically when done.
response_model=schemas.TaskResponse - controls the exact shape of the response. Even if the database object has more fields, only what is defined in the schema goes out. This prevents accidentally leaking sensitive data.
raise HTTPException(status_code=404) - when a record is not found, you raise this and FastAPI returns a proper JSON error response with the correct HTTP status code.
Step 6: main.py
The final file ties everything together.
from fastapi import FastAPI, Request
from fastapi.middleware.cors import CORSMiddleware
import time
import models
from database import engine
from routers import tasks, users
# Create all database tables on startup
models.Base.metadata.create_all(bind=engine)
app = FastAPI(
title="Navas Task Manager API",
description="A simple task manager built with FastAPI and SQLite",
version="1.0.0"
)
# ─── CORS ────────────────────────────────────────
origins = [
"http://localhost:3000", # React dev server
"http://localhost:5173", # Vite dev server
"https://yourfrontend.com", # production frontend
]
app.add_middleware(
CORSMiddleware,
allow_origins=origins,
allow_credentials=True,
allow_methods=["*"],
allow_headers=["*"],
)
# ─── LOGGING MIDDLEWARE ──────────────────────────
@app.middleware("http")
async def log_requests(request: Request, call_next):
start_time = time.time()
print(f"→ {request.method} {request.url}")
response = await call_next(request)
duration = time.time() - start_time
print(f"← {response.status_code} completed in {duration:.3f}s")
return response
# ─── ROUTERS ─────────────────────────────────────
app.include_router(users.router)
app.include_router(tasks.router)
@app.get("/")
def root():
return {"message": "Welcome to Navas Task Manager API "}
Breaking this down:
models.Base.metadata.create_all(bind=engine) - looks at all your model classes and creates the tables in tasks.db if they do not exist yet. Run the app for the first time and the database file appears automatically.
CORS vs custom middleware - two different ways to add middleware:
# Way 1 - for built-in or third-party middleware classes
app.add_middleware(CORSMiddleware, allow_origins=origins, ...)
# Way 2 - for middleware you write yourself
@app.middleware("http")
async def log_requests(request: Request, call_next):
...
Why CORS matters: if your React or Vue frontend runs on localhost:3000 and your API runs on localhost:8000, the browser will block the request by default - because different ports count as different origins. CORS is your API's way of saying "yes, I trust that frontend."
app.include_router() - plugs the routers into the main app. Adding a new resource in the future means writing a new router file and adding one line here.
Running the API
uvicorn main:app --reload
Open your browser and go to:
http://127.0.0.1:8000/docs
You will see the interactive docs with all 10 endpoints grouped by resource, colour-coded by HTTP method. Click any endpoint, hit Execute, and test it live.
Watching the Middleware Work
As you test endpoints in the browser, watch your terminal. Every request prints something like:
→ POST http://127.0.0.1:8000/users/
← 200 completed in 0.038s
→ POST http://127.0.0.1:8000/tasks/
← 200 completed in 0.025s
→ GET http://127.0.0.1:8000/users/
← 200 completed in 0.004s
This is your logging middleware running for every single request — method, URL, status code, and response time. In a production app you would send these logs to a monitoring service. Here they are a clear way to see the request lifecycle in action.
The Nested Response - Where It All Comes Together
Call GET /users/ and look at the response:
[
{
"name": "Alice",
"email": "alice@example.com",
"id": 1,
"tasks": [
{
"title": "Learn FastAPI",
"description": "Build the task manager project",
"id": 1,
"completed": false,
"owner_id": 1
}
]
}
]
The user's tasks are nested inside the response automatically. No extra query was written. This works because:
-
UserResponseschema includestasks: List[TaskResponse] - The
Usermodel hasrelationship("Task", back_populates="owner") -
from_attributes = Truelets Pydantic read the SQLAlchemy object
Three files collaborating silently to produce one clean response.
What To Build Next
Once you are comfortable with this project, here are natural next steps:
Add authentication - hash passwords with passlib and protect endpoints with JWT tokens using python-jose.
Filter by owner - right now GET /tasks/ returns all tasks from all users. Add a query parameter so users only see their own tasks.
Add pagination - add skip and limit parameters to list endpoints so you are not returning the entire database on every call.
Switch to PostgreSQL- change DATABASE_URL from sqlite:///./tasks.db to postgresql://user:password@localhost/dbname and install psycopg2. Everything else stays the same.
Add a due date - add a due_date column using SQLAlchemy's DateTime type and filter tasks by whether they are overdue.
My Final Thoughts
What I love about FastAPI is that the concepts build on each other cleanly. Once you understand get_db as a dependency, the rest of the dependency injection system makes sense. Once you understand one router, you can write any router. Once you write update_task, writing update_user takes two minutes.
That compounding clarity is what makes it a great framework to teach - and to learn.
GitHub Repository
The full project with all files, a clean README, and the .gitignore already configured is available here:
Let's Connect
I write about Data Engineering, AI Engineering, and backend development. If this article helped you, follow me here on dev.to for more.
If you are working on something with FastAPI, Python, or data pipelines and want to collaborate - or just want to talk through an idea - drop a comment below or send me a message. Always happy to connect with people building things.
Built and written by Navas - Data Engineer & AI Engineer
EOF
Top comments (0)