FastAPI is a game-changer for building high-performance APIs in Python. It’s blazing fast, easy to use, and comes with built-in async support. But when you pair it with SQLModel, you’re taking your API development to a whole new level.
SQLModel is like the cool cousin of SQLAlchemy and Pydantic. It brings together the best of both worlds, giving you a powerful ORM with all the type-checking goodness you’d expect. And the best part? It plays really well with FastAPI’s async capabilities.
Let’s dive into how you can integrate SQLModel with FastAPI for some seriously smooth database interactions. Trust me, once you try this combo, you’ll wonder how you ever lived without it.
First things first, we need to set up our environment. Make sure you’ve got Python 3.7+ installed, then run:
pip install fastapi sqlmodel uvicorn
Now, let’s create a simple FastAPI app with SQLModel integration. We’ll build a basic book management system because, well, who doesn’t love books?
from fastapi import FastAPI, HTTPException
from sqlmodel import Field, Session, SQLModel, create_engine, select
from typing import List, Optional
class Book(SQLModel, table=True):
id: Optional[int] = Field(default=None, primary_key=True)
title: str
author: str
year: int
DATABASE_URL = "sqlite:///books.db"
engine = create_engine(DATABASE_URL)
app = FastAPI()
@app.on_event("startup")
def on_startup():
SQLModel.metadata.create_all(engine)
@app.post("/books/", response_model=Book)
def create_book(book: Book):
with Session(engine) as session:
session.add(book)
session.commit()
session.refresh(book)
return book
@app.get("/books/", response_model=List[Book])
def read_books():
with Session(engine) as session:
books = session.exec(select(Book)).all()
return books
@app.get("/books/{book_id}", response_model=Book)
def read_book(book_id: int):
with Session(engine) as session:
book = session.get(Book, book_id)
if not book:
raise HTTPException(status_code=404, detail="Book not found")
return book
@app.put("/books/{book_id}", response_model=Book)
def update_book(book_id: int, book: Book):
with Session(engine) as session:
db_book = session.get(Book, book_id)
if not db_book:
raise HTTPException(status_code=404, detail="Book not found")
book_data = book.dict(exclude_unset=True)
for key, value in book_data.items():
setattr(db_book, key, value)
session.add(db_book)
session.commit()
session.refresh(db_book)
return db_book
@app.delete("/books/{book_id}")
def delete_book(book_id: int):
with Session(engine) as session:
book = session.get(Book, book_id)
if not book:
raise HTTPException(status_code=404, detail="Book not found")
session.delete(book)
session.commit()
return {"ok": True}
This code sets up a basic CRUD (Create, Read, Update, Delete) API for managing books. It’s pretty straightforward, but let’s break it down a bit.
We start by defining our Book model. SQLModel makes this super easy - just inherit from SQLModel and define your fields. The table=True argument tells SQLModel that this model should be mapped to a database table.
Next, we set up our database connection. We’re using SQLite here for simplicity, but you can easily switch this out for PostgreSQL, MySQL, or any other supported database.
The on_startup event handler ensures our database tables are created when the app starts up. No more manual table creation - how cool is that?
Now, let’s look at our API endpoints. We’ve got the full CRUD suite here:
- POST /books/ to create a new book
- GET /books/ to list all books
- GET /books/{book_id} to get a specific book
- PUT /books/{book_id} to update a book
- DELETE /books/{book_id} to delete a book
Each of these endpoints uses SQLModel’s Session to interact with the database. It’s clean, it’s intuitive, and it just works.
But wait, you might be thinking, “This is all synchronous. What about that async support you promised?” Well, my friend, you’re in for a treat. Let’s refactor our app to use async database operations.
from fastapi import FastAPI, HTTPException
from sqlmodel import Field, SQLModel, select
from sqlmodel.ext.asyncio.session import AsyncSession
from typing import List, Optional
from sqlalchemy.ext.asyncio import create_async_engine
from sqlalchemy.orm import sessionmaker
class Book(SQLModel, table=True):
id: Optional[int] = Field(default=None, primary_key=True)
title: str
author: str
year: int
DATABASE_URL = "sqlite+aiosqlite:///books.db"
engine = create_async_engine(DATABASE_URL, echo=True, future=True)
async def get_session() -> AsyncSession:
async_session = sessionmaker(
engine, class_=AsyncSession, expire_on_commit=False
)
async with async_session() as session:
yield session
app = FastAPI()
@app.on_event("startup")
async def on_startup():
async with engine.begin() as conn:
await conn.run_sync(SQLModel.metadata.create_all)
@app.post("/books/", response_model=Book)
async def create_book(book: Book, session: AsyncSession = Depends(get_session)):
session.add(book)
await session.commit()
await session.refresh(book)
return book
@app.get("/books/", response_model=List[Book])
async def read_books(session: AsyncSession = Depends(get_session)):
result = await session.execute(select(Book))
books = result.scalars().all()
return books
@app.get("/books/{book_id}", response_model=Book)
async def read_book(book_id: int, session: AsyncSession = Depends(get_session)):
book = await session.get(Book, book_id)
if not book:
raise HTTPException(status_code=404, detail="Book not found")
return book
@app.put("/books/{book_id}", response_model=Book)
async def update_book(book_id: int, book: Book, session: AsyncSession = Depends(get_session)):
db_book = await session.get(Book, book_id)
if not db_book:
raise HTTPException(status_code=404, detail="Book not found")
book_data = book.dict(exclude_unset=True)
for key, value in book_data.items():
setattr(db_book, key, value)
session.add(db_book)
await session.commit()
await session.refresh(db_book)
return db_book
@app.delete("/books/{book_id}")
async def delete_book(book_id: int, session: AsyncSession = Depends(get_session)):
book = await session.get(Book, book_id)
if not book:
raise HTTPException(status_code=404, detail="Book not found")
await session.delete(book)
await session.commit()
return {"ok": True}
Now we’re cooking with gas! This version of our app uses async database operations throughout. We’re using aiosqlite as our database driver, which allows for async SQLite operations.
The get_session function is now an async generator that yields an AsyncSession. We use FastAPI’s dependency injection system to provide this session to our route handlers.
All of our database operations are now prefixed with await. This allows our app to handle other requests while waiting for database operations to complete, potentially increasing our throughput significantly.
But we’re not done yet. Let’s add some more advanced features to really show off what FastAPI and SQLModel can do together.
First, let’s add pagination to our book list endpoint:
from fastapi import Query
@app.get("/books/", response_model=List[Book])
async def read_books(
skip: int = Query(0, ge=0),
limit: int = Query(10, ge=1, le=100),
session: AsyncSession = Depends(get_session)
):
result = await session.execute(select(Book).offset(skip).limit(limit))
books = result.scalars().all()
return books
Now users can specify how many books they want to see and which page they’re on. This is crucial for performance when dealing with large datasets.
Next, let’s add some filtering capabilities:
@app.get("/books/", response_model=List[Book])
async def read_books(
skip: int = Query(0, ge=0),
limit: int = Query(10, ge=1, le=100),
title: Optional[str] = None,
author: Optional[str] = None,
year: Optional[int] = None,
session: AsyncSession = Depends(get_session)
):
query = select(Book)
if title:
query = query.where(Book.title.contains(title))
if author:
query = query.where(Book.author.contains(author))
if year:
query = query.where(Book.year == year)
result = await session.execute(query.offset(skip).limit(limit))
books = result.scalars().all()
return books
Now users can filter books by title, author, or year. The power of SQLModel really shines here - it’s so easy to build complex queries!
Let’s add one more feature: a search endpoint that looks for matches in both title and author:
@app.get("/books/search/", response_model=List[Book])
async def search_books(
q: str = Query(..., min_length=3),
session: AsyncSession = Depends(get_session)
):
query = select(Book).where(
or_(
Book.title.contains(q),
Book.author.contains(q)
)
)
result = await session.execute(query)
books = result.scalars().all()
return books
This endpoint allows users to search for books with a single query parameter that matches against both title and author.
Now, let’s talk about some best practices when using FastAPI with SQLModel:
-
Always use async operations when possible. They can significantly improve your app’s performance.
-
Use dependency injection for your database sessions. This ensures that sessions are properly managed and closed after each request.
-
Use SQLModel’s built-in validation. It can save you a lot of headaches down the line.
-
Be mindful of N+1 query problems. SQLModel makes it easy to create efficient queries, so use that power!
-
Use transactions for operations that modify multiple records. SQLModel’s Session objects support the async context manager protocol, making this easy.
-
Consider using background tasks for operations that don’t need to block the response.
-
Don’t forget about database migrations! While SQLModel can create tables for you, in a production environment you’ll want more control over schema changes.
Speaking of migrations, let’s add Alembic to our project for database migrations:
pip install alembic
alembic init alembic
Now, edit your alembic/env.py file to use your SQLModel metadata:
from sqlmodel import SQLModel
from your_app import Book # import all your models
target_metadata = SQLModel.metadata
You can now use Alembic to manage your database schema changes:
alembic revision --autogenerate -m "Create books table"
alembic upgrade head
This setup allows you to evolve your database schema over time without losing data.
One last thing: testing. FastAPI makes it super easy to test your API. Here’s a quick example using pytest