PYTHON FASTAPI · LESSON 5 OF 8

Persist tasks with SQLite and dependencies

Keep tasks after restarts and manage a database connection per request.

SQLite
Depends
yield
lifespan

Your runnable checkpoint

Extract into a new practice folder. Each download includes complete source, requirements.txt and README.md. Stop any earlier server before running this example.

Download lesson source ↓

Create and activate a virtual environment as shown in lesson 1. Install with python -m pip install -r requirements.txt, then run fastapi dev main.py. Lessons 6–8 require TASK_API_KEY as explained in lesson 6.

1. Understand the concepts

Now replace the process dictionary with a SQLite file. Python includes sqlite3, so no database server is needed. Start the server in the same directory each time; the default tasks.db path is relative to that directory.

lifespan runs initialization before the application accepts traffic. CREATE TABLE IF NOT EXISTS creates the table on a fresh installation. It does not migrate existing schemas; real schema changes require a migration plan.

Depends(get_db) opens a connection for a request. yield lends it to the route, and finally closes it even after an error. check_same_thread=False permits FastAPI thread-pool usage; it does not remove SQLite write-locking limits. Each request receives its own connection.

Use ? placeholders and separate values for SQL parameters. Never interpolate user input into SQL strings. commit makes a write durable. sqlite3.Row lets us convert named columns into dictionaries; the response model converts the stored 0/1 done flag to a JSON boolean.

The sync def endpoints are appropriate for this blocking database driver. Do not move blocking database calls into async def merely to make the code look asynchronous. For this small demo, startup performs one short schema setup before serving requests.

2. Build and run

These are complete files, not fragments. Save them beside each other in the extracted project folder. Open http://127.0.0.1:8000/docs to send requests.

main.py

import os
import sqlite3
from contextlib import asynccontextmanager
from typing import Annotated
from fastapi import Depends, FastAPI, HTTPException, Query, Response
from pydantic import BaseModel, ConfigDict, Field

DB_PATH = os.getenv("TASK_DB", "tasks.db")

def get_db():
    connection = sqlite3.connect(DB_PATH, check_same_thread=False)
    connection.row_factory = sqlite3.Row
    try:
        yield connection
    finally:
        connection.close()

@asynccontextmanager
async def lifespan(app: FastAPI):
    connection = sqlite3.connect(DB_PATH)
    try:
        connection.execute("CREATE TABLE IF NOT EXISTS tasks (id INTEGER PRIMARY KEY, title TEXT NOT NULL, done INTEGER NOT NULL DEFAULT 0)")
        connection.commit()
    finally:
        connection.close()
    yield

app = FastAPI(title="EduPoly Task API", lifespan=lifespan)
Database = Annotated[sqlite3.Connection, Depends(get_db)]

class TaskCreate(BaseModel):
    model_config = ConfigDict(str_strip_whitespace=True, extra="forbid")
    title: str = Field(min_length=1, max_length=120)
    done: bool = False

class TaskRead(TaskCreate):
    id: int

def find_task(db, task_id):
    row = db.execute("SELECT * FROM tasks WHERE id = ?", (task_id,)).fetchone()
    if row is None:
        raise HTTPException(status_code=404, detail="Task not found")
    return dict(row)

@app.get("/health")
def health():
    return {"status": "ok"}

@app.post("/tasks", response_model=TaskRead, status_code=201)
def create_task(task: TaskCreate, db: Database):
    cursor = db.execute("INSERT INTO tasks (title, done) VALUES (?, ?)", (task.title, task.done))
    db.commit()
    return find_task(db, cursor.lastrowid)

@app.get("/tasks", response_model=list[TaskRead])
def list_tasks(db: Database, limit: Annotated[int, Query(ge=1, le=100)] = 20):
    return [dict(row) for row in db.execute("SELECT * FROM tasks ORDER BY id LIMIT ?", (limit,))]

@app.get("/tasks/{task_id}", response_model=TaskRead)
def read_task(task_id: int, db: Database):
    return find_task(db, task_id)

@app.put("/tasks/{task_id}", response_model=TaskRead)
def replace_task(task_id: int, task: TaskCreate, db: Database):
    find_task(db, task_id)
    db.execute("UPDATE tasks SET title = ?, done = ? WHERE id = ?", (task.title, task.done, task_id))
    db.commit()
    return find_task(db, task_id)

@app.delete("/tasks/{task_id}", status_code=204)
def delete_task(task_id: int, db: Database):
    find_task(db, task_id)
    db.execute("DELETE FROM tasks WHERE id = ?", (task_id,))
    db.commit()
    return Response(status_code=204)

requirements.txt

fastapi[standard]>=0.115,<1
pydantic>=2,<3
httpx>=0.27,<1
pytest>=8,<10

3. Check the result

Create a task, stop the server, restart from the same folder, and fetch the task.
Create a title containing an apostrophe; it must save correctly.
Confirm a missing id returns 404 and an invalid body returns 422.

4. Practice independently

Add an offset query parameter for pagination.

Show the explanation

Validate offset with Query(ge=0), then use LIMIT ? OFFSET ? with (limit, offset). Keep ORDER BY id for deterministic ordering.

When something goes wrong

For import errors, confirm the virtual environment is active and install requirements.txt using that environment’s Python. For 422, read the response detail and compare your input to the model. For 404, check the method, path and resource id. For 401 in later lessons, use Authorize in /docs with the same key configured on the server.