FastAPI SQLAlchemy实战：数据库操作与API开发 – wiki词典

作者： /

FastAPI SQLAlchemy实战：数据库操作与API开发

在现代Web开发中，构建高性能、易于维护的API至关重要。FastAPI以其卓越的性能和现代Python特性（如类型提示）脱颖而出，成为构建RESTful API的首选框架之一。当与强大的Python ORM（对象关系映射）工具SQLAlchemy结合时，数据库操作变得更加优雅和高效。

本文将通过一个实战示例，详细介绍如何使用FastAPI和SQLAlchemy进行数据库操作和API开发，包括环境搭建、模型定义、CRUD操作以及会话管理。

1. 简介

  • FastAPI: 一个现代、快速（高性能）的Web框架，用于使用Python 3.7+构建API。它基于标准的Python类型提示，内置数据验证、序列化和交互式API文档（Swagger UI/ReDoc）。
  • SQLAlchemy: Python SQL工具包和ORM，提供了全功能的持久化框架，可以用于与各种数据库进行交互。它允许开发者使用Python对象来操作数据库，而不是直接编写SQL语句。

结合两者，我们可以构建出类型安全、文档完善、性能优异的API应用。

2. 环境搭建与项目初始化

首先，我们需要创建一个项目目录并安装必要的依赖。

“`bash
mkdir fastapi_sql_app
cd fastapi_sql_app
python -m venv venv
./venv/Scripts/activate # Windows

source venv/bin/activate # macOS/Linux

pip install fastapi uvicorn “SQLAlchemy[asyncio]” “sqlite-sync” “pydantic[email]”

注意：SQLAlchemy 2.0+ 推荐使用异步驱动，这里我们为了演示使用同步驱动的sqlite-sync。

如果使用PostgreSQL，可以安装 psycopg2-binary；如果使用MySQL，可以安装 pymysql。

“`

在项目根目录下创建一个 main.py 文件，这将是我们的主应用文件。

3. 数据库配置与会话管理

我们将创建一个 database.py 文件来处理数据库的连接和会话管理。为了简化，我们将使用SQLite数据库。

database.py

“`python
from sqlalchemy import create_engine
from sqlalchemy.orm import sessionmaker, declarative_base

数据库连接URL

sqlite:///./sql_app.db 表示在当前目录下创建一个名为 sql_app.db 的 SQLite 数据库文件

SQLALCHEMY_DATABASE_URL = “sqlite:///./sql_app.db”

创建 SQLAlchemy 引擎

connect_args={“check_same_thread”: False} 是 SQLite 数据库特有的配置，

允许多个线程访问同一个连接，对于 FastAPI 来说是必要的，因为它可能在多个线程中处理请求。

engine = create_engine(
SQLALCHEMY_DATABASE_URL, connect_args={“check_same_thread”: False}
)

创建一个 SessionLocal 类，每个请求都会创建一个独立的数据库会话

autocommit=False 意味着我们不会自动提交事务，需要手动提交

autoflush=False 意味着我们不会在查询前自动刷新会话

SessionLocal = sessionmaker(autocommit=False, autoflush=False, bind=engine)

创建一个 Base 类，将来我们的所有 SQLAlchemy 模型都会继承它

Base = declarative_base()

数据库会话依赖项

def get_db():
db = SessionLocal()
try:
yield db
finally:
db.close()
“`

4. 定义SQLAlchemy模型

接下来，我们定义数据库表结构。创建一个 models.py 文件。

models.py

“`python
from sqlalchemy import Column, Integer, String, Boolean
from sqlalchemy.orm import relationship

from .database import Base

class User(Base):
tablename = “users” # 表名

id = Column(Integer, primary_key=True, index=True)
email = Column(String, unique=True, index=True)
hashed_password = Column(String)
is_active = Column(Boolean, default=True)

items = relationship("Item", back_populates="owner") # 定义与 Item 模型的关系

class Item(Base):
tablename = “items” # 表名

id = Column(Integer, primary_key=True, index=True)
title = Column(String, index=True)
description = Column(String, index=True)
owner_id = Column(Integer, foreign_key=True) # 外键，关联到 users 表的 id

owner = relationship("User", back_populates="items") # 定义与 User 模型的关系

“`

5. 定义Pydantic Schema

Pydantic用于定义数据模型、验证请求体和响应数据。创建一个 schemas.py 文件。

schemas.py

“`python
from pydantic import BaseModel

class ItemBase(BaseModel):
title: str
description: str | None = None

class ItemCreate(ItemBase):
pass

class Item(ItemBase):
id: int
owner_id: int

class Config:
    orm_mode = True # 告诉 Pydantic 将 SQLAlchemy 模型转换为 Pydantic 模型

class UserBase(BaseModel):
email: str

class UserCreate(UserBase):
password: str

class User(UserBase):
id: int
is_active: bool
items: list[Item] = [] # 用户的物品列表

class Config:
    orm_mode = True

“`

6. CRUD操作（业务逻辑）

我们将把CRUD操作封装在 crud.py 文件中，使其与API路由分离，提高代码可读性和可维护性。

crud.py

“`python
from sqlalchemy.orm import Session

from . import models, schemas

def get_user(db: Session, user_id: int):
return db.query(models.User).filter(models.User.id == user_id).first()

def get_user_by_email(db: Session, email: str):
return db.query(models.User).filter(models.User.email == email).first()

def get_users(db: Session, skip: int = 0, limit: int = 100):
return db.query(models.User).offset(skip).limit(limit).all()

def create_user(db: Session, user: schemas.UserCreate):
fake_hashed_password = user.password + “notreallyhashed”
db_user = models.User(email=user.email, hashed_password=fake_hashed_password)
db.add(db_user)
db.commit()
db.refresh(db_user)
return db_user

def get_items(db: Session, skip: int = 0, limit: int = 100):
return db.query(models.Item).offset(skip).limit(limit).all()

def create_user_item(db: Session, item: schemas.ItemCreate, user_id: int):
db_item = models.Item(**item.model_dump(), owner_id=user_id)
db.add(db_item)
db.commit()
db.refresh(db_item)
return db_item

def update_item(db: Session, item_id: int, item: schemas.ItemCreate):
db_item = db.query(models.Item).filter(models.Item.id == item_id).first()
if db_item:
for key, value in item.model_dump().items():
setattr(db_item, key, value)
db.commit()
db.refresh(db_item)
return db_item

def delete_item(db: Session, item_id: int):
db_item = db.query(models.Item).filter(models.Item.id == item_id).first()
if db_item:
db.delete(db_item)
db.commit()
return db_item
“`

7. 创建FastAPI路由

现在，我们将所有组件整合到 main.py 中，定义API端点。

main.py

“`python
from fastapi import FastAPI, Depends, HTTPException, status
from sqlalchemy.orm import Session

from . import crud, models, schemas
from .database import SessionLocal, engine

创建数据库表

models.Base.metadata.create_all(bind=engine)

app = FastAPI()

依赖项，获取数据库会话

def get_db():
db = SessionLocal()
try:
yield db
finally:
db.close()

@app.post(“/users/”, response_model=schemas.User)
def create_user(user: schemas.UserCreate, db: Session = Depends(get_db)):
db_user = crud.get_user_by_email(db, email=user.email)
if db_user:
raise HTTPException(status_code=status.HTTP_400_BAD_REQUEST, detail=”Email already registered”)
return crud.create_user(db=db, user=user)

@app.get(“/users/”, response_model=list[schemas.User])
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.User)
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=status.HTTP_404_NOT_FOUND, detail=”User not found”)
return db_user

@app.post(“/users/{user_id}/items/”, response_model=schemas.Item)
def create_item_for_user(
user_id: int, item: schemas.ItemCreate, db: Session = Depends(get_db)
):
return crud.create_user_item(db=db, item=item, user_id=user_id)

@app.get(“/items/”, response_model=list[schemas.Item])
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

@app.put(“/items/{item_id}”, response_model=schemas.Item)
def update_item(item_id: int, item: schemas.ItemCreate, db: Session = Depends(get_db)):
db_item = crud.update_item(db, item_id, item)
if db_item is None:
raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail=”Item not found”)
return db_item

@app.delete(“/items/{item_id}”, status_code=status.HTTP_204_NO_CONTENT)
def delete_item(item_id: int, db: Session = Depends(get_db)):
db_item = crud.delete_item(db, item_id)
if db_item is None:
raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail=”Item not found”)
return {“message”: “Item deleted successfully”}
“`

8. 运行应用程序

在项目根目录下，使用Uvicorn运行FastAPI应用：

bash
uvicorn main:app --reload

现在，打开浏览器访问 http://127.0.0.1:8000/docs，你将看到由Swagger UI生成的交互式API文档。你可以直接在文档中测试各个端点：

  • 创建用户: POST /users/
  • 获取用户列表: GET /users/
  • 获取单个用户: GET /users/{user_id}
  • 为用户创建物品: POST /users/{user_id}/items/
  • 获取物品列表: GET /items/
  • 更新物品: PUT /items/{item_id}
  • 删除物品: DELETE /items/{item_id}

9. 总结

本文详细介绍了如何将FastAPI与SQLAlchemy结合，从零开始构建一个具备CRUD功能的API应用。我们涵盖了以下关键点：

  1. 项目结构: 清晰地组织 database.py, models.py, schemas.py, crud.py 和 main.py 文件。
  2. 数据库连接: 配置SQLAlchemy引擎和会话管理。
  3. ORM模型: 使用declarative_base定义SQLAlchemy模型，映射数据库表结构。
  4. 数据验证: 利用Pydantic定义请求和响应数据模型，实现自动验证和序列化。
  5. 业务逻辑: 将CRUD操作封装在crud.py中，保持API路由的简洁。
  6. API路由: 在FastAPI中定义HTTP端点，利用Depends进行依赖注入（如数据库会话）。
  7. 交互式文档: FastAPI自动生成的Swagger UI使得API测试和理解变得轻而易举。

通过这个实战示例，你应该对如何在FastAPI应用中高效地使用SQLAlchemy进行数据库操作有了深入的理解。在此基础上，你可以进一步探索更高级的主题，如认证授权、异步数据库操作（使用SQLAlchemy[asyncio]和asyncio兼容的数据库驱动）、数据库迁移工具（如Alembic）等。

滚动至顶部