Connect FastAPI to SQLite

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.db means the database file will be named tasks.db and saved in the same folder as database.py.
• check_same_thread=False is required because FastAPI uses multiple threads.
• SessionLocal is a factory that creates database sessions.
• Base is 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 Column defines a field: id (integer, primary key), title (string), description (string), completed (boolean).
• index=True makes 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_db dependency 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

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.