Kritim Yantra
Apr 14, 2025
In the previous post, we created a simple FastAPI To-Do app using in-memory storage and TDD. Now, let's add a real database using SQLite so we can persist data even after restarting the server.
We will continue to follow Test-Driven Development (TDD), which means writing tests before writing the actual implementation.
pip install fastapi uvicorn pytest httpx sqlalchemy
fastapi-todo-tdd/
โโโ app/
โ โโโ __init__.py
โ โโโ main.py
โ โโโ database.py
โ โโโ models.py
โโโ tests/
โ โโโ test_main.py
โโโ requirements.txt
โโโ README.md
from sqlalchemy import create_engine
from sqlalchemy.orm import sessionmaker, declarative_base
SQLALCHEMY_DATABASE_URL = "sqlite:///./todos.db"
engine = create_engine(
SQLALCHEMY_DATABASE_URL, connect_args={"check_same_thread": False}
)
SessionLocal = sessionmaker(autocommit=False, autoflush=False, bind=engine)
Base = declarative_base()
Explanation: We configure SQLite using SQLAlchemy. The check_same_thread=False
is required for SQLite to allow multiple threads.
from sqlalchemy import Column, Integer, String
from app.database import Base
class ToDo(Base):
__tablename__ = "todos"
id = Column(Integer, primary_key=True, index=True)
title = Column(String, index=True)
Explanation: We define a ToDo
model/table with two fields: id
and title
. SQLAlchemy will use this to create the todos
table.
from fastapi import FastAPI, Depends, HTTPException, status
from sqlalchemy.orm import Session
from pydantic import BaseModel
from typing import List
from app import models, database
from app.database import SessionLocal, engine
models.Base.metadata.create_all(bind=engine)
app = FastAPI()
def get_db():
db = SessionLocal()
try:
yield db
finally:
db.close()
class ToDoCreate(BaseModel):
title: str
class ToDoResponse(ToDoCreate):
id: int
class Config:
orm_mode = True
@app.post("/todos", status_code=status.HTTP_201_CREATED, response_model=ToDoResponse)
def create_todo(todo: ToDoCreate, db: Session = Depends(get_db)):
db_todo = models.ToDo(title=todo.title)
db.add(db_todo)
db.commit()
db.refresh(db_todo)
return db_todo
@app.get("/todos", response_model=List[ToDoResponse])
def get_todos(db: Session = Depends(get_db)):
return db.query(models.ToDo).all()
@app.put("/todos/{todo_id}", response_model=ToDoResponse)
def update_todo(todo_id: int, updated: ToDoCreate, db: Session = Depends(get_db)):
todo = db.query(models.ToDo).filter(models.ToDo.id == todo_id).first()
if not todo:
raise HTTPException(status_code=404, detail="ToDo not found")
todo.title = updated.title
db.commit()
db.refresh(todo)
return todo
@app.delete("/todos/{todo_id}", status_code=204)
def delete_todo(todo_id: int, db: Session = Depends(get_db)):
todo = db.query(models.ToDo).filter(models.ToDo.id == todo_id).first()
if not todo:
raise HTTPException(status_code=404, detail="ToDo not found")
db.delete(todo)
db.commit()
return
Explanation:
POST /todos
creates a new to-do.GET /todos
lists all to-dos.PUT /todos/{id}
updates a to-do.DELETE /todos/{id}
removes a to-do.from fastapi.testclient import TestClient
from app.main import app
client = TestClient(app)
def test_create_todo():
response = client.post("/todos", json={"title": "Learn TDD"})
assert response.status_code == 201
assert response.json()["title"] == "Learn TDD"
def test_get_todos():
response = client.get("/todos")
assert response.status_code == 200
assert isinstance(response.json(), list)
def test_update_todo():
create = client.post("/todos", json={"title": "Temp"})
todo_id = create.json()["id"]
update = client.put(f"/todos/{todo_id}", json={"title": "Updated Title"})
assert update.status_code == 200
assert update.json()["title"] == "Updated Title"
def test_delete_todo():
create = client.post("/todos", json={"title": "To Delete"})
todo_id = create.json()["id"]
delete = client.delete(f"/todos/{todo_id}")
assert delete.status_code == 204
Explanation:
Each function tests one endpoint. TDD means we should write these before we implement each endpoint.
You can improve testing by using an in-memory SQLite database. Update your test_main.py
to override the DB:
# tests/base.py
from sqlalchemy import create_engine
from sqlalchemy.orm import sessionmaker
from sqlalchemy.pool import StaticPool
from app.database import Base
from app.main import app, get_db
from fastapi.testclient import TestClient
# one in-memory DB, shared across all sessions
SQLALCHEMY_DATABASE_URL = "sqlite:///:memory:"
engine = create_engine(
SQLALCHEMY_DATABASE_URL,
connect_args={"check_same_thread": False},
poolclass=StaticPool,
)
TestingSessionLocal = sessionmaker(autocommit=False, autoflush=False, bind=engine)
# create tables once, on that single connection
Base.metadata.create_all(bind=engine)
def override_get_db():
db = TestingSessionLocal()
try:
yield db
finally:
db.close()
# wire up the override before TestClient is instantiated
app.dependency_overrides[get_db] = override_get_db
client = TestClient(app)
Explanation: We override the default DB connection with an in-memory SQLite for isolated, faster test runs.
Happy coding โจ
No comments yet. Be the first to comment!
Please log in to post a comment:
Sign in with Google