SQLModel + Neon Setup

Your Task API currently stores data in a Python list. Restart the server—everything's gone. Real APIs need persistent storage. This lesson connects your FastAPI app to Neon PostgreSQL using SQLModel.

Why this combination?

• SQLModel = Pydantic + SQLAlchemy in one. Your models work for both validation AND database.
• Neon = PostgreSQL as a service. No installation, free tier, instant setup.

This is the "fast track" approach. No migrations, no complex setup. Get persistent data working, then add complexity later if needed.

Setting Up Neon

Step 1: Go to neon.tech and create a free account

Step 2: Create a new project (default settings are fine)

Step 3: Copy your connection string:

postgresql://user:password@ep-xxx.region.aws.neon.tech/neondb?sslmode=require

Step 4: Add to your .env:

DATABASE_URL=postgresql://user:password@ep-xxx.region.aws.neon.tech/neondb?sslmode=require

That's it. You have a PostgreSQL database.

Installing SQLModel

uv add sqlmodel psycopg2-binary

• sqlmodel - The ORM that combines Pydantic and SQLAlchemy
• psycopg2-binary - PostgreSQL driver

Defining Your Model

SQLModel classes with table=True become database tables. Create models.py:

from sqlmodel import SQLModel, Field
from typing import Optional
from datetime import datetime


class Task(SQLModel, table=True):
    """Task stored in database."""
    id: Optional[int] = Field(default=None, primary_key=True)
    title: str = Field(min_length=1, max_length=200)
    description: Optional[str] = None
    status: str = Field(default="pending")
    created_at: datetime = Field(default_factory=datetime.utcnow)

What each part means:

• table=True - This model maps to a database table
• primary_key=True - Auto-incrementing ID
• Field(default=...) - Column defaults
• Optional[int] = None - ID is None until database assigns it

Connecting to the Database

Create database.py:

from sqlmodel import SQLModel, create_engine, Session
from config import get_settings

settings = get_settings()

engine = create_engine(settings.database_url, echo=True)


def create_db_and_tables():
    """Create all tables in the database."""
    SQLModel.metadata.create_all(engine)


def get_session():
    """Dependency that provides database sessions."""
    with Session(engine) as session:
        yield session

Key concepts:

• create_engine() - Establishes connection to Neon
• echo=True - Logs SQL queries (helpful for debugging)
• Session - Context for database operations
• yield session - Provides session to endpoint, closes when done

Creating Tables on Startup

Call create_db_and_tables() when the app starts. Update your main.py:

from fastapi import FastAPI
from database import create_db_and_tables

app = FastAPI()


@app.on_event("startup")
def on_startup():
    create_db_and_tables()

First time you run the app, the table is created in Neon.

CRUD Operations with Session

Create

from fastapi import Depends
from sqlmodel import Session
from models import Task
from database import get_session


@app.post("/tasks", status_code=201)
def create_task(
    task: Task,
    session: Session = Depends(get_session)
):
    session.add(task)
    session.commit()
    session.refresh(task)  # Get the assigned ID
    return task

Output:

{
  "id": 1,
  "title": "Learn SQLModel",
  "description": null,
  "status": "pending",
  "created_at": "2024-01-15T10:30:00"
}

Read (List)

from sqlmodel import select


@app.get("/tasks")
def list_tasks(session: Session = Depends(get_session)):
    tasks = session.exec(select(Task)).all()
    return tasks

Read (Single)

from fastapi import HTTPException


@app.get("/tasks/{task_id}")
def get_task(task_id: int, session: Session = Depends(get_session)):
    task = session.get(Task, task_id)
    if not task:
        raise HTTPException(status_code=404, detail="Task not found")
    return task

Update

@app.put("/tasks/{task_id}")
def update_task(
    task_id: int,
    task_update: Task,
    session: Session = Depends(get_session)
):
    task = session.get(Task, task_id)
    if not task:
        raise HTTPException(status_code=404, detail="Task not found")

    task.title = task_update.title
    task.description = task_update.description
    task.status = task_update.status

    session.add(task)
    session.commit()
    session.refresh(task)
    return task

Delete

@app.delete("/tasks/{task_id}", status_code=204)
def delete_task(task_id: int, session: Session = Depends(get_session)):
    task = session.get(Task, task_id)
    if not task:
        raise HTTPException(status_code=404, detail="Task not found")

    session.delete(task)
    session.commit()
    return None

Complete Example

Here's everything together. Create models.py:

from sqlmodel import SQLModel, Field
from typing import Optional
from datetime import datetime


class Task(SQLModel, table=True):
    id: Optional[int] = Field(default=None, primary_key=True)
    title: str = Field(min_length=1, max_length=200)
    description: Optional[str] = None
    status: str = Field(default="pending")
    created_at: datetime = Field(default_factory=datetime.utcnow)

Create database.py:

from sqlmodel import SQLModel, create_engine, Session
from config import get_settings

settings = get_settings()
engine = create_engine(settings.database_url, echo=False)


def create_db_and_tables():
    SQLModel.metadata.create_all(engine)


def get_session():
    with Session(engine) as session:
        yield session

Create main.py:

from fastapi import FastAPI, Depends, HTTPException
from sqlmodel import Session, select
from models import Task
from database import create_db_and_tables, get_session

app = FastAPI(title="Task API")


@app.on_event("startup")
def on_startup():
    create_db_and_tables()


@app.post("/tasks", status_code=201)
def create_task(task: Task, session: Session = Depends(get_session)):
    session.add(task)
    session.commit()
    session.refresh(task)
    return task


@app.get("/tasks")
def list_tasks(session: Session = Depends(get_session)):
    return session.exec(select(Task)).all()


@app.get("/tasks/{task_id}")
def get_task(task_id: int, session: Session = Depends(get_session)):
    task = session.get(Task, task_id)
    if not task:
        raise HTTPException(status_code=404, detail="Task not found")
    return task


@app.put("/tasks/{task_id}")
def update_task(
    task_id: int,
    task_update: Task,
    session: Session = Depends(get_session)
):
    task = session.get(Task, task_id)
    if not task:
        raise HTTPException(status_code=404, detail="Task not found")

    task.title = task_update.title
    task.description = task_update.description
    task.status = task_update.status

    session.add(task)
    session.commit()
    session.refresh(task)
    return task


@app.delete("/tasks/{task_id}", status_code=204)
def delete_task(task_id: int, session: Session = Depends(get_session)):
    task = session.get(Task, task_id)
    if not task:
        raise HTTPException(status_code=404, detail="Task not found")
    session.delete(task)
    session.commit()

Output:

$ curl -X POST http://localhost:8000/tasks \
  -H "Content-Type: application/json" \
  -d '{"title": "Learn SQLModel"}'

{"id":1,"title":"Learn SQLModel","description":null,"status":"pending","created_at":"2024-01-15T10:30:00"}

$ curl http://localhost:8000/tasks
[{"id":1,"title":"Learn SQLModel","description":null,"status":"pending","created_at":"2024-01-15T10:30:00"}]

Restart the server. The data persists. That's the difference a database makes.

Hands-On Exercise

Step 1: Create a Neon account and get your connection string

Step 2: Add DATABASE_URL to your .env

Step 3: Create models.py and database.py as shown above

Step 4: Update main.py to use the database

Step 5: Test CRUD operations through Swagger UI

Step 6: Restart the server and verify data persists

Common Mistakes

Mistake 1: Forgetting table=True

# Wrong - just a Pydantic model, no database table
class Task(SQLModel):
    title: str

# Correct - creates database table
class Task(SQLModel, table=True):
    title: str

Mistake 2: Missing primary key

# Wrong - no primary key defined
class Task(SQLModel, table=True):
    title: str

# Correct - has primary key
class Task(SQLModel, table=True):
    id: Optional[int] = Field(default=None, primary_key=True)
    title: str

Mistake 3: Not calling session.commit()

# Wrong - changes not saved
session.add(task)
return task  # ID is still None!

# Correct - commit saves to database
session.add(task)
session.commit()
session.refresh(task)  # Now has ID
return task

Mistake 4: Using wrong connection string

# Wrong - missing sslmode for Neon
DATABASE_URL=postgresql://user:pass@host/db

# Correct - SSL required for Neon
DATABASE_URL=postgresql://user:pass@host/db?sslmode=require

What About Migrations?

This "fast track" approach uses create_all() which:

• Creates tables that don't exist
• Does NOT modify existing tables

If you change your model (add a column), you need to either:

1. Drop and recreate the table (loses data)
2. Use Alembic migrations (covered in Chapter 47)

For learning, the fast track is fine. For production, you'll add migrations.

Try With AI

After completing the exercise, explore these scenarios.

Prompt 1: Filtering Queries

I have tasks in my database. How do I filter them?
For example, get only tasks with status="pending"
or tasks created in the last 7 days.

Show me SQLModel select() with where() clauses.

What you're learning: Select queries can filter, sort, and limit results. This is essential for real-world APIs.

Prompt 2: Separate Request/Response Models

My Task model has table=True for the database, but I want
different fields for API requests and responses.
For example, I don't want clients to set created_at.

How do I separate database models from API models in SQLModel?

What you're learning: Real APIs often have TaskCreate, TaskRead, and Task (database) as separate models. SQLModel supports this pattern.

Prompt 3: Connection Pooling

My API might get many concurrent requests. Each request
creates a new database connection with get_session().
Is this efficient? Should I use connection pooling?

What you're learning: Production databases use connection pools. SQLAlchemy (which SQLModel uses) has pooling built in—understanding when it matters is important.

---

Reflect on Your Skill

You built a fastapi-agent skill in Lesson 0. Test and improve it based on what you learned.

Test Your Skill

Using my fastapi-agent skill, help me set up database integration with SQLModel and Neon.
Does my skill include model definitions, create_engine, Session management, and CRUD operations?

Identify Gaps

Ask yourself:

• Did my skill include SQLModel table definitions with table=True?
• Did it handle database connection setup and Session dependency injection?
• Did it cover CRUD operations using Session (add, exec, get, commit, refresh)?

Improve Your Skill

If you found gaps:

My fastapi-agent skill is missing database integration patterns.
Update it to include SQLModel table definitions, Neon connection setup,
Session management with yield dependencies, and proper CRUD operations.