Introduction: Why Do We Need a Database?
Chapter 17: Connecting FastAPI to SQLite – Making Your Data Persistent
What is SQLite and Why Use It?
So far, our Tasks API stores all tasks in a Python list. When you restart the server, all data disappears. This is like writing on a whiteboard and then erasing it. A database solves this by storing data permanently on your computer's hard drive.
SQLite is a lightweight, file-based database. It doesn't need a separate server or complex setup. It's perfect for learning, small projects, and even for production mobile apps. Python comes with built-in support for SQLite, so you don't need to install anything extra.
Why Do We Need a Database for Our API?
- Persistence: Data survives server restarts.
- Structure: Data is organized in tables with rows and columns.
- Querying: You can easily search, filter, and update data using SQL.
- Scalability: A database can handle much more data than a list.
Step 1: Install the Database Driver
FastAPI works with many databases. For SQLite, we'll use a library called sqlalchemy (which is an ORM – Object Relational Mapper) and aiosqlite for async support. Open your terminal in the project folder and run:
pip install sqlalchemy aiosqlite
Step 2: Create the Database Configuration File
Inside your project folder (where main.py lives), create a new file called database.py. This file will handle the connection to SQLite.
# database.py
from sqlalchemy import create_engine
from sqlalchemy.ext.declarative import declarative_base
from sqlalchemy.orm import sessionmaker
# SQLite database URL – the file will be created in your project folder
SQLALCHEMY_DATABASE_URL = "sqlite:///./tasks.db"
# Create the engine (the connection to the database)
engine = create_engine(
SQLALCHEMY_DATABASE_URL,
connect_args={"check_same_thread": False} # Needed for SQLite
)
# Create a session local class to interact with the database
SessionLocal = sessionmaker(autocommit=False, autoflush=False, bind=engine)
# Base class for our models (tables)
Base = declarative_base()
Explanation:
sqlite:///./tasks.dbmeans the database file will be namedtasks.dband saved in the same folder asdatabase.py.check_same_thread=Falseis required because FastAPI uses multiple threads.SessionLocalis a factory that creates database sessions.Baseis the foundation for all our table models.
Step 3: Define the Task Model (Table)
Now, create a file called models.py. This will define what a "task" looks like in the database.
# models.py
from sqlalchemy import Column, Integer, String, Boolean
from database import Base
class Task(Base):
__tablename__ = "tasks"
id = Column(Integer, primary_key=True, index=True)
title = Column(String, index=True)
description = Column(String, default="")
completed = Column(Boolean, default=False)
Explanation:
__tablename__sets the name of the table in the database.- Each
Columndefines a field:id(integer, primary key),title(string),description(string),completed(boolean). index=Truemakes searching faster.
Step 4: Create the Database Tables
Open your main.py and add the following code at the top (after imports) to create the tables when the app starts:
# main.py (add these lines)
from fastapi import FastAPI
from database import engine, Base
from models import Task # import the model
# Create all tables in the database
Base.metadata.create_all(bind=engine)
app = FastAPI()
When you run the app, SQLite will create the tasks.db file and the tasks table automatically.
Step 5: Add a Database Session Dependency
We need a way to get a fresh database session for each request. Create a new file dependencies.py:
# dependencies.py
from database import SessionLocal
def get_db():
db = SessionLocal()
try:
yield db
finally:
db.close()
This function creates a session, gives it to the endpoint, and closes it when the request finishes.
Step 6: Rewrite the CRUD Endpoints to Use the Database
Now, let's update main.py to use the database instead of a list. Here's a complete example for creating and listing tasks:
# main.py (updated)
from fastapi import FastAPI, Depends, HTTPException
from sqlalchemy.orm import Session
from database import engine, Base
from models import Task
from dependencies import get_db
from pydantic import BaseModel
# Create tables
Base.metadata.create_all(bind=engine)
app = FastAPI()
# Pydantic schema for request/response
class TaskCreate(BaseModel):
title: str
description: str = ""
completed: bool = False
class TaskResponse(TaskCreate):
id: int
class Config:
orm_mode = True
# Create a new task
@app.post("/tasks/", response_model=TaskResponse)
def create_task(task: TaskCreate, db: Session = Depends(get_db)):
db_task = Task(**task.dict())
db.add(db_task)
db.commit()
db.refresh(db_task)
return db_task
# Get all tasks
@app.get("/tasks/", response_model=list[TaskResponse])
def read_tasks(db: Session = Depends(get_db)):
tasks = db.query(Task).all()
return tasks
# Get a single task by ID
@app.get("/tasks/{task_id}", response_model=TaskResponse)
def read_task(task_id: int, db: Session = Depends(get_db)):
task = db.query(Task).filter(Task.id == task_id).first()
if task is None:
raise HTTPException(status_code=404, detail="Task not found")
return task
Key changes:
- We use
db: Session = Depends(get_db)to get a database session. db.add()adds a new record,db.commit()saves it,db.refresh()gets the auto-generated ID.db.query(Task).all()fetches all tasks.db.query(Task).filter(Task.id == task_id).first()finds one task by ID.
Common Mistakes and How to Avoid Them
- Forgetting to commit: Always call
db.commit()after adding or updating data, otherwise changes are lost. - Not closing the session: The
get_dbdependency handles this automatically, but if you create sessions manually, always close them. - Wrong column names: Make sure the column names in your model match the field names in your Pydantic schema.
- Missing
orm_mode = True: Without this, Pydantic cannot convert SQLAlchemy objects to JSON.
Practical Callout: Testing Your Database-Powered API
Try it now:
- Run your FastAPI server:
uvicorn main:app --reload - Open Swagger UI at
http://127.0.0.1:8000/docs - Create a few tasks using the POST endpoint.
- Stop the server (Ctrl+C) and restart it.
- Use the GET endpoint – your tasks are still there! That's persistence.
Practice Task
Add two more endpoints to your main.py:
- PUT /tasks/{task_id} – Update an existing task (title, description, or completed status).
- DELETE /tasks/{task_id} – Delete a task by its ID.
Hint: For update, use db.query(Task).filter(Task.id == task_id).first() to find the task, then update its fields and call db.commit(). For delete, use db.delete(task) then db.commit().
Test both endpoints using Swagger UI. If you get stuck, review the patterns used in the GET and POST examples above.
Loading ratings...