> ## Documentation Index
> Fetch the complete documentation index at: https://fastapi2day.codewithsiva.dev/llms.txt
> Use this file to discover all available pages before exploring further.

# SQL & ORM Integration with FastAPI and SQLModel Guide

> Replace in-memory storage with a real SQL database in FastAPI using SQLModel and SQLAlchemy — models, sessions, CRUD operations, and migrations.

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.

```mermaid theme={null}
graph TD
    SQLModel --> SQLAlchemy[SQLAlchemy: Database Tables & Queries]
    SQLModel --> Pydantic[Pydantic: Data Validation & Serialization]
    SQLModel --> FastAPI[FastAPI: Automatic API docs & request handling]
```

***

## Installation

```bash theme={null}
pip install sqlmodel
```

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

```bash theme={null}
pip install psycopg2-binary   # PostgreSQL driver
```

***

## Defining Database Models

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

```python theme={null}
# 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:

```python theme={null}
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

```python theme={null}
# 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

```python theme={null}
# 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

```python theme={null}
# 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()`                             |

<Note>
  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`.
</Note>

***

## Filtering and Querying

SQLModel's `select()` supports rich query composition:

```python theme={null}
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

<Tabs>
  <Tab title="SQLite (Development)">
    ```python theme={null}
    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.
  </Tab>

  <Tab title="PostgreSQL (Production)">
    ```python theme={null}
    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:

    ```bash theme={null}
    pip install psycopg2-binary
    ```
  </Tab>
</Tabs>

<Tip>
  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.
</Tip>

***

## Updated Project Structure

```text theme={null}
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.