SQL & ORM Integration with FastAPI and SQLModel Guide

Up to this point your Employee Management System has stored data in a Python dictionary that disappears every time the server restarts. In production, you need a persistent database. This lesson shows you how to connect FastAPI to a real SQL database using SQLModel — a library built on top of SQLAlchemy and Pydantic that lets your data models serve dual duty as both database table definitions and API schema validators. By the end, you’ll have a fully persistent CRUD API backed by a SQL database.

Why SQLModel?

FastAPI’s creator also built SQLModel specifically to work alongside FastAPI. It combines:

SQLAlchemy — the industry-standard Python SQL toolkit and ORM
Pydantic — the validation library FastAPI already uses

This means a single class can define both the database table structure and the Pydantic validation schema, eliminating the duplication of maintaining separate ORM and Pydantic models.

Installation

pip install sqlmodel

SQLModel installs SQLAlchemy and Pydantic automatically as dependencies. For PostgreSQL, also install:

pip install psycopg2-binary   # PostgreSQL driver

Defining Database Models

With SQLModel, table=True tells SQLAlchemy to create an actual database table for this class:

# models.py
from typing import Optional
from sqlmodel import SQLModel, Field

class Employee(SQLModel, table=True):
    id: Optional[int] = Field(default=None, primary_key=True)
    name: str = Field(min_length=2, max_length=100)
    department: str
    role: str

The id field is Optional because the database generates it automatically on insert — you don’t provide it.

Request and Response Schemas

Define separate schemas without table=True for request validation and response filtering:

class EmployeeCreate(SQLModel):
    name: str = Field(min_length=2)
    department: str
    role: str

class EmployeeRead(SQLModel):
    id: int
    name: str
    department: str
    role: str

class EmployeeUpdate(SQLModel):
    name: Optional[str] = None
    department: Optional[str] = None
    role: Optional[str] = None

Database Connection and Session Management

# database.py
from sqlmodel import create_engine, Session, SQLModel

# SQLite for development (no server needed)
DATABASE_URL = "sqlite:///./employees.db"

# PostgreSQL for production:
# DATABASE_URL = "postgresql://user:password@localhost:5432/ems_db"

engine = create_engine(DATABASE_URL, echo=True)

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

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

The get_session function is a generator dependency — it opens a session, yields it to the endpoint, and automatically closes it when the request is done (even if an exception is raised).

Wiring Up the Application

# main.py
from contextlib import asynccontextmanager
from fastapi import FastAPI
from database import create_db_and_tables
from routers import employees

@asynccontextmanager
async def lifespan(app: FastAPI):
    create_db_and_tables()   # Create tables on startup
    yield

app = FastAPI(title="EMS API", lifespan=lifespan)
app.include_router(employees.router)

The lifespan context manager runs create_db_and_tables() once when the application starts, creating any missing database tables.

CRUD Endpoints with a Real Database

# routers/employees.py
from typing import Annotated
from fastapi import APIRouter, Depends, HTTPException, status
from sqlmodel import Session, select
from database import get_session
from models import Employee, EmployeeCreate, EmployeeRead, EmployeeUpdate

router = APIRouter(prefix="/employees", tags=["Employees"])

SessionDep = Annotated[Session, Depends(get_session)]


# READ ALL
@router.get("/", response_model=list[EmployeeRead])
def list_employees(
    session: SessionDep,
    skip: int = 0,
    limit: int = 10
):
    employees = session.exec(select(Employee).offset(skip).limit(limit)).all()
    return employees


# READ ONE
@router.get("/{employee_id}", response_model=EmployeeRead)
def get_employee(employee_id: int, session: SessionDep):
    employee = session.get(Employee, employee_id)
    if not employee:
        raise HTTPException(
            status_code=status.HTTP_404_NOT_FOUND,
            detail=f"Employee {employee_id} not found"
        )
    return employee


# CREATE
@router.post("/", response_model=EmployeeRead, status_code=status.HTTP_201_CREATED)
def create_employee(employee_data: EmployeeCreate, session: SessionDep):
    employee = Employee.model_validate(employee_data)
    session.add(employee)
    session.commit()
    session.refresh(employee)   # reload to get database-generated fields like id
    return employee


# UPDATE
@router.put("/{employee_id}", response_model=EmployeeRead)
def update_employee(
    employee_id: int,
    employee_data: EmployeeUpdate,
    session: SessionDep
):
    employee = session.get(Employee, employee_id)
    if not employee:
        raise HTTPException(
            status_code=status.HTTP_404_NOT_FOUND,
            detail=f"Employee {employee_id} not found"
        )

    update_data = employee_data.model_dump(exclude_unset=True)
    for field, value in update_data.items():
        setattr(employee, field, value)

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


# DELETE
@router.delete("/{employee_id}", status_code=status.HTTP_204_NO_CONTENT)
def delete_employee(employee_id: int, session: SessionDep):
    employee = session.get(Employee, employee_id)
    if not employee:
        raise HTTPException(
            status_code=status.HTTP_404_NOT_FOUND,
            detail=f"Employee {employee_id} not found"
        )
    session.delete(employee)
    session.commit()

Key Database Operations

Operation	SQLModel Code
Insert a record	`session.add(obj)` → `session.commit()` → `session.refresh(obj)`
Fetch by primary key	`session.get(Model, id)`
Fetch with conditions	`session.exec(select(Model).where(Model.field == value)).all()`
Update a record	`setattr(obj, field, value)` → `session.add(obj)` → `session.commit()`
Delete a record	`session.delete(obj)` → `session.commit()`

Always call session.refresh(employee) after a commit() when you need to return the object. The commit() flushes changes to the database but may clear the in-memory object. refresh() reloads the latest data from the database, including auto-generated fields like id and created_at.

Filtering and Querying

SQLModel’s select() supports rich query composition:

from sqlmodel import select

# Filter by department
statement = select(Employee).where(Employee.department == "Engineering")

# Search with LIKE
statement = select(Employee).where(Employee.name.contains("Alice"))

# Order and paginate
statement = select(Employee).order_by(Employee.name).offset(0).limit(20)

# Execute
employees = session.exec(statement).all()

SQLite vs PostgreSQL

SQLite (Development)
PostgreSQL (Production)

DATABASE_URL = "sqlite:///./employees.db"
engine = create_engine(DATABASE_URL, echo=True)

SQLite requires no separate server and stores the database in a single file. Ideal for local development and testing.

DATABASE_URL = "postgresql://user:password@localhost:5432/ems_db"
engine = create_engine(DATABASE_URL, pool_size=10, max_overflow=20)

PostgreSQL is the recommended database for production. Install the driver with:

pip install psycopg2-binary

Use environment variables for database credentials in production. Never hard-code passwords in your source code. Libraries like python-dotenv or FastAPI’s Settings pattern (using Pydantic BaseSettings) make this easy.

Updated Project Structure

app/
├── main.py
├── database.py          ← engine, session, create_tables
├── models.py            ← SQLModel table definitions + schemas
└── routers/
    ├── __init__.py
    └── employees.py     ← CRUD endpoints using Session

Switching from in-memory to database storage required changes only to database.py, models.py, and the repository layer — the Router’s interface stayed exactly the same. This is the payoff of the layered architecture you built in the previous lesson.

SQL & Databases

Data Analysis

FastAPI

Frontend

SQL & ORM Integration with FastAPI and SQLModel Guide

Why SQLModel?

Installation

Defining Database Models

Request and Response Schemas

Database Connection and Session Management

Wiring Up the Application

CRUD Endpoints with a Real Database

Key Database Operations

Filtering and Querying

SQLite vs PostgreSQL

Updated Project Structure

​Why SQLModel?

​Installation

​Defining Database Models

​Request and Response Schemas

​Database Connection and Session Management

​Wiring Up the Application

​CRUD Endpoints with a Real Database

​Key Database Operations

​Filtering and Querying

​SQLite vs PostgreSQL

​Updated Project Structure

Why SQLModel?

Installation

Defining Database Models

Request and Response Schemas

Database Connection and Session Management

Wiring Up the Application

CRUD Endpoints with a Real Database

Key Database Operations

Filtering and Querying

SQLite vs PostgreSQL

Updated Project Structure