FastAPI+SQLAlchemy：快速构建数据库驱动的API

为什么选择FastAPI+SQLAlchemy？¶

在Python Web开发中，FastAPI以其高性能、自动生成API文档和简洁的语法受到广泛欢迎；而SQLAlchemy作为强大的ORM（对象关系映射）工具，能让开发者用Python类操作数据库，避免直接编写复杂SQL语句。两者结合可以快速构建出既高效又易于维护的数据库驱动API，非常适合初学者入门。

准备工作：安装依赖¶

首先，确保你的Python环境已安装。打开终端，执行以下命令安装所需库：

pip install fastapi uvicorn sqlalchemy pydantic

FastAPI：核心Web框架，用于构建API
Uvicorn：ASGI服务器，用于运行FastAPI应用
SQLAlchemy：ORM工具，处理数据库交互
Pydantic：数据验证和序列化库（FastAPI依赖它处理请求/响应数据）

项目结构设计¶

我们采用模块化结构，将不同功能拆分到独立文件中，便于维护：

myapp/
├── main.py          # FastAPI主程序，定义API路由
├── database.py      # 数据库连接配置
├── models.py        # SQLAlchemy ORM模型（数据库表结构）
├── schemas.py       # Pydantic模型（数据验证/序列化）
└── crud.py          # CRUD操作（增删改查）

1. 数据库配置（database.py）¶

这部分负责创建数据库连接和会话管理。我们以SQLite为例（无需额外服务器，适合初学者）：

from sqlalchemy import create_engine
from sqlalchemy.ext.declarative import declarative_base
from sqlalchemy.orm import sessionmaker

# 数据库连接URL（SQLite无需额外配置，文件路径即为数据库）
SQLALCHEMY_DATABASE_URL = "sqlite:///./test.db"

# 创建SQLAlchemy引擎（数据库连接入口）
engine = create_engine(
    SQLALCHEMY_DATABASE_URL, connect_args={"check_same_thread": False}  # SQLite多线程兼容
)

# 创建会话本地类（用于每次请求获取独立会话）
SessionLocal = sessionmaker(autocommit=False, autoflush=False, bind=engine)

# 基础模型类（所有ORM模型需继承此类）
Base = declarative_base()

2. 数据库模型（models.py）¶

使用SQLAlchemy定义数据库表结构（ORM模型），以用户表为例：

from sqlalchemy import Column, Integer, String
from database import Base

class User(Base):
    __tablename__ = "users"  # 数据库表名

    id = Column(Integer, primary_key=True, index=True)  # 主键
    name = Column(String, index=True)                   # 姓名（索引优化查询）
    email = Column(String, unique=True, index=True)     # 邮箱（唯一，索引）
    age = Column(Integer, nullable=True)                # 年龄（可选）

说明：
- Base 继承自SQLAlchemy的declarative_base，是所有模型的父类
- __tablename__ 指定数据库表名
- Column 定义表字段，需指定类型（如Integer、String）和约束（如主键、唯一）

3. Pydantic模型（schemas.py）¶

FastAPI用Pydantic定义数据验证和序列化规则（区分请求和响应格式）：

from pydantic import BaseModel

# 用于创建用户的请求数据验证（不含ID，ID由数据库自动生成）
class UserCreate(BaseModel):
    name: str
    email: str
    age: int | None = None  # 可选字段

# 用于返回用户数据的响应格式（包含所有字段）
class UserResponse(BaseModel):
    id: int
    name: str
    email: str
    age: int | None = None

    class Config:
        orm_mode = True  # 允许从ORM模型（SQLAlchemy对象）创建实例

说明：
- UserCreate 用于接收前端提交的数据（如POST请求），验证数据合法性
- UserResponse 用于返回查询结果，orm_mode=True 允许直接从SQLAlchemy模型转换数据

4. CRUD操作（crud.py）¶

实现数据库的增删改查（Create, Read, Update, Delete）：

from sqlalchemy.orm import Session
from models import User
from schemas import UserCreate, UserResponse

# 创建用户
def create_user(db: Session, user: UserCreate) -> UserResponse:
    db_user = User(name=user.name, email=user.email, age=user.age)
    db.add(db_user)
    db.commit()       # 提交事务
    db.refresh(db_user)  # 刷新对象（获取数据库生成的ID）
    return db_user

# 获取单个用户
def get_user(db: Session, user_id: int) -> UserResponse | None:
    return db.query(User).filter(User.id == user_id).first()

# 获取用户列表
def get_users(db: Session, skip: int = 0, limit: int = 100) -> list[UserResponse]:
    return db.query(User).offset(skip).limit(limit).all()

# 更新用户
def update_user(db: Session, user_id: int, user_update: UserCreate) -> UserResponse | None:
    db_user = db.query(User).filter(User.id == user_id).first()
    if not db_user:
        return None
    # 更新字段
    db_user.name = user_update.name
    db_user.email = user_update.email
    db_user.age = user_update.age
    db.commit()
    db.refresh(db_user)
    return db_user

# 删除用户
def delete_user(db: Session, user_id: int) -> bool:
    db_user = db.query(User).filter(User.id == user_id).first()
    if not db_user:
        return False
    db.delete(db_user)
    db.commit()
    return True

5. 主程序（main.py）¶

整合所有模块，定义API路由：

from fastapi import Depends, FastAPI, HTTPException
from sqlalchemy.orm import Session
from database import engine, SessionLocal, Base
from models import User
from schemas import UserCreate, UserResponse
from crud import create_user, get_user, get_users, update_user, delete_user

# 创建数据库表（首次运行时执行，需导入所有模型）
Base.metadata.create_all(bind=engine)

app = FastAPI(title="用户API示例")

# 依赖项：获取数据库会话（每次请求自动创建/关闭会话）
def get_db():
    db = SessionLocal()
    try:
        yield db
    finally:
        db.close()  # 确保会话关闭，避免连接泄露

# API路由
@app.post("/users/", response_model=UserResponse)
def create_new_user(user: UserCreate, db: Session = Depends(get_db)):
    """创建新用户"""
    return create_user(db=db, user=user)

@app.get("/users/{user_id}", response_model=UserResponse | None)
def read_user(user_id: int, db: Session = Depends(get_db)):
    """获取单个用户"""
    db_user = get_user(db=db, user_id=user_id)
    if db_user is None:
        raise HTTPException(status_code=404, detail="用户不存在")
    return db_user

@app.get("/users/", response_model=list[UserResponse])
def read_users(skip: int = 0, limit: int = 10, db: Session = Depends(get_db)):
    """获取用户列表"""
    users = get_users(db=db, skip=skip, limit=limit)
    return users

@app.put("/users/{user_id}", response_model=UserResponse | None)
def update_existing_user(user_id: int, user: UserCreate, db: Session = Depends(get_db)):
    """更新用户信息"""
    return update_user(db=db, user_id=user_id, user_update=user)

@app.delete("/users/{user_id}", response_model=bool)
def delete_existing_user(user_id: int, db: Session = Depends(get_db)):
    """删除用户"""
    success = delete_user(db=db, user_id=user_id)
    if not success:
        raise HTTPException(status_code=404, detail="用户不存在")
    return success

运行与测试¶

启动应用：在终端执行以下命令：

   uvicorn main:app --reload

main:app 表示从main.py文件导入app实例
--reload 开启热重载，代码修改后自动重启

测试API：
- 访问Swagger UI：http://localhost:8000/docs，可交互式测试所有接口
- 或用curl命令测试：

     # 创建用户
     curl -X POST "http://localhost:8000/users/" -H "Content-Type: application/json" -d '{"name":"Alice","email":"alice@example.com","age":25}'

     # 获取用户列表
     curl "http://localhost:8000/users/"

核心优势总结¶

FastAPI：自动生成API文档、支持异步、数据验证强
SQLAlchemy：ORM简化数据库操作，无需手写SQL
模块化设计：分离路由、模型、验证、业务逻辑，便于扩展
依赖注入：自动管理数据库会话，避免连接泄露

通过以上步骤，你已完成一个基础的数据库驱动API开发。接下来可尝试扩展功能，如添加用户权限、实现多表关联（如订单表与用户表）等。实践是学习编程的最佳方式，动手修改代码，探索更多FastAPI+SQLAlchemy的可能性吧！