FastAPI Database Integration
FastAPI can be integrated with various databases, and the most common approach is to use SQLAlchemy as the ORM. This section introduces how to build an API with CRUD (Create, Read, Update, Delete) functionality using FastAPI + SQLAlchemy.
Install dependencies
pip install sqlalchemy
Project Structure
project/ ├── main.py # FastAPI 应用入口 ├── database.py # 数据库连接配置 ├── models.py # SQLAlchemy 模型 ├── schemas.py # Pydantic 数据模型 └── crud.py # 数据库操作函数
1. Database Configuration
Indatabase.pyConfigure the database connection in:
Example
# database.py
from sqlalchemy import create_engine
from sqlalchemy.orm import sessionmaker, DeclarativeBase
# SQLite database (suitable for development)
SQLALCHEMY_DATABASE_URL = "sqlite:///./sql_app.db"
# Create database engine
# connect_args is only needed for SQLite, allowing multi-threaded access
engine = create_engine(
SQLALCHEMY_DATABASE_URL,
connect_args={"check_same_thread": False}
)
# Create session factory
SessionLocal = sessionmaker(autocommit=False, autoflush=False, bind=engine)
# Declare Base Class
class Base(DeclarativeBase):
pass
from sqlalchemy import create_engine
from sqlalchemy.orm import sessionmaker, DeclarativeBase
# SQLite database (suitable for development)
SQLALCHEMY_DATABASE_URL = "sqlite:///./sql_app.db"
# Create database engine
# connect_args is only needed for SQLite, allowing multi-threaded access
engine = create_engine(
SQLALCHEMY_DATABASE_URL,
connect_args={"check_same_thread": False}
)
# Create session factory
SessionLocal = sessionmaker(autocommit=False, autoflush=False, bind=engine)
# Declare Base Class
class Base(DeclarativeBase):
pass
For the development environment, SQLite is sufficient. It is a file-based database that requires no additional database service installation. For the production environment, PostgreSQL or MySQL is recommended; you only need to modify the
SQLALCHEMY_DATABASE_URLThat's it.
2. SQLAlchemy Model
Inmodels.pyDefine the database table structure in:
Example
# models.py
from sqlalchemy import Column, Integer, String
from database import Base
class User(Base):
__tablename__ = "users" # Table name
id = Column(Integer, primary_key=True, index=True) # Primary key, auto-increment
email = Column(String, unique=True, index=True) # Unique, create index
hashed_password = Column(String) # Hash Password
is_active = Column(Boolean, default=True) # Is Active
class Item(Base):
__tablename__ = "items"
id = Column(Integer, primary_key=True, index=True)
title = Column(String, index=True) # Title, create index
description = Column(String) # Description
owner_id = Column(Integer, ForeignKey("users.id")) # Foreign key references the users table
from sqlalchemy import Column, Integer, String
from database import Base
class User(Base):
__tablename__ = "users" # Table name
id = Column(Integer, primary_key=True, index=True) # Primary key, auto-increment
email = Column(String, unique=True, index=True) # Unique, create index
hashed_password = Column(String) # Hash Password
is_active = Column(Boolean, default=True) # Is Active
class Item(Base):
__tablename__ = "items"
id = Column(Integer, primary_key=True, index=True)
title = Column(String, index=True) # Title, create index
description = Column(String) # Description
owner_id = Column(Integer, ForeignKey("users.id")) # Foreign key references the users table
3. Pydantic Data Models
Inschemas.pydefining the input/output data structures of the API:
Example
# schemas.py
from pydantic import BaseModel
# ===== User Model =====
class UserBase(BaseModel):
email: str
class UserCreate(UserBase):
password: str # Password required when creating
class UserOut(UserBase):
id: int
is_active: bool
model_config = ConfigDict(from_attributes=True) # Supports ORM objects
# ===== Product Model =====
class ItemBase(BaseModel):
title: str
description: str | None = None
class ItemCreate(ItemBase):
pass
class ItemOut(ItemBase):
id: int
owner_id: int
model_config = ConfigDict(from_attributes=True)
from pydantic import BaseModel
# ===== User Model =====
class UserBase(BaseModel):
email: str
class UserCreate(UserBase):
password: str # Password required when creating
class UserOut(UserBase):
id: int
is_active: bool
model_config = ConfigDict(from_attributes=True) # Supports ORM objects
# ===== Product Model =====
class ItemBase(BaseModel):
title: str
description: str | None = None
class ItemCreate(ItemBase):
pass
class ItemOut(ItemBase):
id: int
owner_id: int
model_config = ConfigDict(from_attributes=True)
Pydantic models and SQLAlchemy models are separate. Pydantic is responsible for data validation at the API layer, while SQLAlchemy is responsible for table structures at the database layer.
from_attributes=TrueEnable Pydantic to read data from SQLAlchemy objects.
4. Database Operation Functions
Incrud.pyEncapsulate database operations in:
Example
# crud.py
from sqlalchemy.orm import Session
import models, schemas
def get_user(db: Session, user_id: int):
"""Get user by ID"""
return db.query(models.User).filter(models.User.id == user_id).first()
def get_user_by_email(db: Session, email: str):
"""Get user by email"""
return db.query(models.User).filter(models.User.email == email).first()
def get_users(db: Session, skip: int = 0, limit: int = 100):
"""Get user list"""
return db.query(models.User).offset(skip).limit(limit).all()
def create_user(db: Session, user: schemas.UserCreate):
"""Create user"""
fake_hashed_password = user.password + "notreallyhashed" # In practice, use passlib for hashing
db_user = models.User(
email=user.email,
hashed_password=fake_hashed_password
)
db.add(db_user)
db.commit() # Commit Transaction
db.refresh(db_user) # Refresh the object to get the database-generated id
return db_user
def get_items(db: Session, skip: int = 0, limit: int = 100):
"""Get product list"""
return db.query(models.Item).offset(skip).limit(limit).all()
def create_item(db: Session, item: schemas.ItemCreate, user_id: int):
"""Create product"""
db_item = models.Item(**item.model_dump(), owner_id=user_id)
db.add(db_item)
db.commit()
db.refresh(db_item)
return db_item
from sqlalchemy.orm import Session
import models, schemas
def get_user(db: Session, user_id: int):
"""Get user by ID"""
return db.query(models.User).filter(models.User.id == user_id).first()
def get_user_by_email(db: Session, email: str):
"""Get user by email"""
return db.query(models.User).filter(models.User.email == email).first()
def get_users(db: Session, skip: int = 0, limit: int = 100):
"""Get user list"""
return db.query(models.User).offset(skip).limit(limit).all()
def create_user(db: Session, user: schemas.UserCreate):
"""Create user"""
fake_hashed_password = user.password + "notreallyhashed" # In practice, use passlib for hashing
db_user = models.User(
email=user.email,
hashed_password=fake_hashed_password
)
db.add(db_user)
db.commit() # Commit Transaction
db.refresh(db_user) # Refresh the object to get the database-generated id
return db_user
def get_items(db: Session, skip: int = 0, limit: int = 100):
"""Get product list"""
return db.query(models.Item).offset(skip).limit(limit).all()
def create_item(db: Session, item: schemas.ItemCreate, user_id: int):
"""Create product"""
db_item = models.Item(**item.model_dump(), owner_id=user_id)
db.add(db_item)
db.commit()
db.refresh(db_item)
return db_item
5. FastAPI Application Entry Point
Inmain.pyIntegrate all components in:
Example
# main.py
from typing import Annotated
from fastapi import Depends, FastAPI, HTTPException
from sqlalchemy.orm import Session
from pydantic import ConfigDict
import models, schemas, crud
from database import engine, SessionLocal, Base
# Create database table
Base.metadata.create_all(bind=engine)
app = FastAPI()
# Dependency: get database session
def get_db():
db = SessionLocal()
try:
yield db
finally:
db.close()
# ===== User Routes =====
@app.post("/users/", response_model=schemas.UserOut)
def create_user(user: schemas.UserCreate, db: Session = Depends(get_db)):
# Check whether the email already exists
db_user = crud.get_user_by_email(db, email=user.email)
if db_user:
raise HTTPException(status_code=400, detail=“Email already registered”)
return crud.create_user(db=db, user=user)
@app.get("/users/", response_model=list[schemas.UserOut])
def read_users(skip: int = 0, limit: int = 100, db: Session = Depends(get_db)):
users = crud.get_users(db, skip=skip, limit=limit)
return users
@app.get("/users/{user_id}", response_model=schemas.UserOut)
def read_user(user_id: int, db: Session = Depends(get_db)):
db_user = crud.get_user(db, user_id=user_id)
if db_user is None:
raise HTTPException(status_code=404, detail=“User does not exist”)
return db_user
# ===== Product Routes =====
@app.post("/users/{user_id}/items/", response_model=schemas.ItemOut)
def create_item_for_user(
user_id: int,
item: schemas.ItemCreate,
db: Session = Depends(get_db),
):
return crud.create_item(db=db, item=item, user_id=user_id)
@app.get("/items/", response_model=list[schemas.ItemOut])
def read_items(skip: int = 0, limit: int = 100, db: Session = Depends(get_db)):
items = crud.get_items(db, skip=skip, limit=limit)
return items
from typing import Annotated
from fastapi import Depends, FastAPI, HTTPException
from sqlalchemy.orm import Session
from pydantic import ConfigDict
import models, schemas, crud
from database import engine, SessionLocal, Base
# Create database table
Base.metadata.create_all(bind=engine)
app = FastAPI()
# Dependency: get database session
def get_db():
db = SessionLocal()
try:
yield db
finally:
db.close()
# ===== User Routes =====
@app.post("/users/", response_model=schemas.UserOut)
def create_user(user: schemas.UserCreate, db: Session = Depends(get_db)):
# Check whether the email already exists
db_user = crud.get_user_by_email(db, email=user.email)
if db_user:
raise HTTPException(status_code=400, detail=“Email already registered”)
return crud.create_user(db=db, user=user)
@app.get("/users/", response_model=list[schemas.UserOut])
def read_users(skip: int = 0, limit: int = 100, db: Session = Depends(get_db)):
users = crud.get_users(db, skip=skip, limit=limit)
return users
@app.get("/users/{user_id}", response_model=schemas.UserOut)
def read_user(user_id: int, db: Session = Depends(get_db)):
db_user = crud.get_user(db, user_id=user_id)
if db_user is None:
raise HTTPException(status_code=404, detail=“User does not exist”)
return db_user
# ===== Product Routes =====
@app.post("/users/{user_id}/items/", response_model=schemas.ItemOut)
def create_item_for_user(
user_id: int,
item: schemas.ItemCreate,
db: Session = Depends(get_db),
):
return crud.create_item(db=db, item=item, user_id=user_id)
@app.get("/items/", response_model=list[schemas.ItemOut])
def read_items(skip: int = 0, limit: int = 100, db: Session = Depends(get_db)):
items = crud.get_items(db, skip=skip, limit=limit)
return items
Dependency Injection for Database Sessions
The core pattern is to use dependency injection to manage database sessions:
def get_db():
db = SessionLocal() # 创建会话
try:
yield db # 提供会话给路由函数
finally:
db.close() # 请求完成后关闭会话
Each request obtains an independent database session, which is automatically closed after the request completes, avoiding resource leaks.
Connection Strings for Different Databases
| database | Connection string example |
|---|---|
| SQLite | sqlite:///./sql_app.db |
| PostgreSQL | postgresql://user:password@localhost/dbname |
| MySQL | mysql+pymysql://user:password@localhost/dbname |
Switching databases only requires modifying the connection string; the rest of the code does not need to be changed. This is one of the advantages of using an ORM.
Summary
- Use SQLAlchemy as the ORM, and FastAPI manages database sessions through dependency injection.
- Pydantic model (
schemas.py) handles API data validation, SQLAlchemy models (models.py) is responsible for the database table structure - CRUD operations are encapsulated in
crud.pyin, keeping the code clean - For development, use SQLite; for production, switch to PostgreSQL/MySQL by only changing the connection string.
from_attributes=TrueEnable Pydantic to read data from ORM objects.