Using a Database with FastAPI
Reindert-Jan Ekker
@rjekker [Link]
SQLModel
- Built on SQLAlchemy + Pydantic
Overview
Create data model classes
Create a DB connection
CRUD operations
Relations
Transactions
- Session
Object-Relational Mapping
Python Relational DB (SQL)
Classes Tables
Objects Rows
Attributes Columns
Based on SQLAlchemy
- Popular ORM library
- Mature, robust
- Supports many databases
Also based on Pydantic
- Model classes are Pydantic Models
- Allows easy integration with FastAPI
- Same creator (Sebastián Ramírez)
SQLModel
- New, still being developed
- Gives access to the power of SQLAlchemy
- [Link]
We will not use async with SQLModel
Regular functions SQLModel and
FastAPI still runs those concurrently Async
SQLAlchemy async support still beta
Data Model Classes
from sqlmodel import SQLModel, Field
# SQLModel inherits from pydantic BaseModel
# Pass table=True when creating the class to map this to a DB table
class Car(SQLModel, table=True):
id: int | None = Field(primary_key=True, default=None)
start: int
end: int
description: str
Database Setup
from sqlmodel import SQLModel, create_engine
engine = create_engine(
"sqlite:///[Link]",
connect_args={"check_same_thread": False}, # Needed for SQLite
echo=True # Log generated SQL (don’t use in production)
)
# Create the database on startup
@app.on_event("startup")
def on_startup():
[Link].create_all(engine)
Session
from sqlmodel import Session, Depends
def get_session():
with Session(engine) as session:
yield session
# FastAPI will call get_session and store result in session parameter
@[Link](...)
def car_by_id(id: int, session: Session = Depends(get_session)):
car = [Link](Car, id)
if car:
return car
# else return 404
Adding a New Car
@[Link](...)
def add_car(car_input: CarInput,
session: Session = Depends(get_session)) -> Car:
new_car = Car.from_orm(car_input)
[Link](new_car)
[Link]()
[Link](new_car)
return new_car
Querying Cars
from sqlmodel import select
@[Link]("/api/cars")
def get_cars(size: str | None = None, doors: int | None = None,
session: Session = Depends(get_session)) -> list[Car]:
query = select(Car)
if size:
query = [Link]([Link] == size)
if doors:
query = [Link]([Link] >= doors)
return [Link](query).all()
Removing a Car
@[Link]("/api/cars/{id}", status_code=204)
def remove_car(id: int, session: Session = Depends(get_session)) -> None:
car = [Link](Car, id)
if car:
[Link](car)
[Link]()
else:
raise HTTPException(status_code=404)
Relations
class Trip(TripInput, table=True):
id: int | None = Field(default=None, primary_key=True)
car_id: int = Field(foreign_key="[Link]")
car: "Car" = Relationship(back_populates="trips")
class Car(CarInput, table=True):
id: int | None = Field(default=None, primary_key=True)
trips: list[Trip] = Relationship(back_populates="car")
[Link]
[Link]
[Link]
[Link]
SQLModel
- Built on SQLAlchemy + Pydantic
Summary
Create data model classes
Create a DB connection
CRUD operations
Relations
Transactions
- Session