SQL Databases with SQLModel

🎯 What You'll Learn

• Set up SQLModel with FastAPI for database operations
• Create database models with proper field configurations
• Implement comprehensive CRUD operations (Create, Read, Update, Delete)
• Handle database sessions and connections using dependency injection
• Use multiple models for enhanced security and API design
• Configure database engines and startup events

📚 Theory

SQLModel is a library created by the same author as FastAPI, designed specifically to work seamlessly with FastAPI applications that need SQL databases. It combines the power of SQLAlchemy (for database operations) with Pydantic (for data validation).

Essential SQLModel Components

1. Table Model Definition:

from sqlmodel import SQLModel, Field

class Hero(SQLModel, table=True):
    id: int | None = Field(default=None, primary_key=True)
    name: str = Field(index=True)
    age: int | None = Field(default=None, index=True)
    secret_name: str

Key Field Configurations:

• primary_key=True: Marks the field as the table's primary key
• index=True: Creates a database index for faster queries
• default=None: Allows the database to auto-generate values (especially for IDs)

2. Database Engine Setup:

from sqlmodel import create_engine

sqlite_file_name = "database.db"
sqlite_url = f"sqlite:///{sqlite_file_name}"
connect_args = {"check_same_thread": False}
engine = create_engine(sqlite_url, connect_args=connect_args)

3. Database Session Management:

from sqlmodel import Session
from typing import Annotated
from fastapi import Depends

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

SessionDep = Annotated[Session, Depends(get_session)]

4. Database Table Creation:

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

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

CRUD Operations Implementation

Create Operation:

@app.post("/heroes/")
def create_hero(hero: Hero, session: SessionDep) -> Hero:
    session.add(hero)
    session.commit()
    session.refresh(hero)
    return hero

Read Operations:

from sqlmodel import select

# Read multiple with pagination
@app.get("/heroes/")
def read_heroes(
    session: SessionDep, 
    offset: int = 0, 
    limit: Annotated[int, Query(le=100)] = 100
) -> list[Hero]:
    heroes = session.exec(select(Hero).offset(offset).limit(limit)).all()
    return heroes

# Read single by ID
@app.get("/heroes/{hero_id}")
def read_hero(hero_id: int, session: SessionDep) -> Hero:
    hero = session.get(Hero, hero_id)
    if not hero:
        raise HTTPException(status_code=404, detail="Hero not found")
    return hero

Update Operation:

@app.patch("/heroes/{hero_id}")
def update_hero(hero_id: int, hero_update: HeroUpdate, session: SessionDep) -> Hero:
    hero = session.get(Hero, hero_id)
    if not hero:
        raise HTTPException(status_code=404, detail="Hero not found")
    
    hero_data = hero_update.model_dump(exclude_unset=True)
    for key, value in hero_data.items():
        setattr(hero, key, value)
    
    session.add(hero)
    session.commit()
    session.refresh(hero)
    return hero

Delete Operation:

@app.delete("/heroes/{hero_id}")
def delete_hero(hero_id: int, session: SessionDep):
    hero = session.get(Hero, hero_id)
    if not hero:
        raise HTTPException(status_code=404, detail="Hero not found")
    session.delete(hero)
    session.commit()
    return {"ok": True}

🔧 Key Concepts

• SQLModel: Combines SQLAlchemy's database power with Pydantic's validation
• Table Models: Classes with table=True that represent database tables
• Field Configuration: Use Field() to specify primary keys, indexes, and defaults
• Database Engine: Manages database connections and should be created once
• Sessions: Handle individual transactions and should be injected per request
• Dependency Injection: Use Depends() to inject database sessions into endpoints
• Query Builder: Use select() for complex database queries
• Pagination: Implement with offset and limit parameters

💡 Best Practices

• Primary Key Required: Every table model must have a primary key field
• Session Per Request: Use dependency injection for database sessions to ensure thread safety
• Proper Error Handling: Always check if records exist before operations
• Database Startup: Create tables during application startup with @app.on_event("startup")
• Connection Arguments: Use check_same_thread=False for SQLite with FastAPI
• Query Limits: Always limit query results to prevent performance issues
• Transaction Management: Use session.commit() to save changes and session.refresh() to get updated data

🔗 Additional Resources

• SQLModel Documentation
• SQLAlchemy Documentation
• FastAPI SQL Databases Tutorial