FastAPI 与 SQLAlchemy 核心集成教程

欢迎来到 FastAPI 学习教程的 SQLAlchemy 核心集成部分！作为高级工程师，我将指导你如何将 SQLAlchemy——一个强大的 Python ORM（对象关系映射器）与 FastAPI 无缝集成，用于处理数据库操作。本教程假设你已具备基本的 Python 和 FastAPI 知识，适合新手入门。让我们开始吧！

目录

1. SQLAlchemy 简介
2. 为什么在 FastAPI 中使用 SQLAlchemy？
3. 环境准备与安装
4. 设置 SQLAlchemy 核心
5. 定义数据库模型
6. 管理数据库会话
7. 集成到 FastAPI 端点
8. 示例：完整的 CRUD 操作
9. 最佳实践与常见问题
10. 总结与下一步

1. SQLAlchemy 简介

SQLAlchemy 是一个流行的 Python SQL 工具包和 ORM，它允许你使用 Python 对象来操作数据库，而无需直接编写 SQL 语句。它包括两个主要组件：

• SQLAlchemy Core：提供低级的 SQL 抽象，适合需要精细控制的场景。
• SQLAlchemy ORM：基于 Core 构建，提供高级的 ORM 功能。

在本教程中，我们将重点介绍如何集成 SQLAlchemy 核心（包括 ORM），因为 FastAPI 通常与 ORM 结合使用以简化开发。

2. 为什么在 FastAPI 中使用 SQLAlchemy？

• 高效开发：使用 Python 类定义表结构，减少 SQL 代码。
• 类型安全：与 FastAPI 的类型提示（如 Pydantic）配合良好，提高代码可读性和维护性。
• 灵活性：支持多种数据库后端（如 PostgreSQL, MySQL, SQLite）。
• 社区支持：拥有丰富的资源和活跃的社区。

3. 环境准备与安装

确保你已安装 Python（推荐 3.8+）和 pip。然后，安装必要的包：

pip install fastapi sqlalchemy uvicorn

• fastapi: FastAPI 框架。
• sqlalchemy: SQLAlchemy 库。
• uvicorn: ASGI 服务器，用于运行 FastAPI 应用。

如果你计划使用特定数据库（如 PostgreSQL），还需安装相应的驱动程序，例如 pip install psycopg2。

4. 设置 SQLAlchemy 核心

首先，创建一个项目目录（例如 fastapi_sqlalchemy_tutorial），并在其中创建 main.py 文件。我们将从设置 SQLAlchemy 引擎和会话工厂开始。

步骤 1：导入必要的模块

from sqlalchemy import create_engine, Column, Integer, String
from sqlalchemy.ext.declarative import declarative_base
from sqlalchemy.orm import sessionmaker
from fastapi import FastAPI, Depends, HTTPException
from pydantic import BaseModel
from typing import List, Optional

步骤 2：配置数据库连接

创建数据库引擎和会话工厂。这里以 SQLite 为例（易于测试），但你可以替换为其他数据库。

# 数据库 URL：使用 SQLite 内存数据库（适合开发测试）
SQLALCHEMY_DATABASE_URL = "sqlite:///./test.db"  # 或使用 "sqlite:///:memory:" 用于内存数据库

# 创建引擎
engine = create_engine(SQLALCHEMY_DATABASE_URL, connect_args={"check_same_thread": False} if "sqlite" in SQLALCHEMY_DATABASE_URL else {})

# 创建会话工厂
SessionLocal = sessionmaker(autocommit=False, autoflush=False, bind=engine)

# 声明基类用于定义模型
Base = declarative_base()

• create_engine: 连接到数据库的核心函数。
• SessionLocal: 用于创建数据库会话的工厂类。
• Base: 所有模型的基类。

5. 定义数据库模型

使用 SQLAlchemy 的 declarative_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)  # 邮箱，唯一且带索引

然后，创建表（通常在应用启动时执行）：

Base.metadata.create_all(bind=engine)  # 根据模型创建表（如果不存在）

6. 管理数据库会话

在 FastAPI 中，使用依赖注入来管理数据库会话的生命周期。这有助于避免资源泄漏和确保事务正确性。

创建一个依赖函数，用于在每个请求中获取数据库会话，并在请求结束后关闭它。

def get_db():
    """
    依赖函数：提供数据库会话。
    使用 FastAPI 的 Depends() 在端点中调用。
    """
    db = SessionLocal()  # 创建新会话
    try:
        yield db  # 将会话传递给端点
    finally:
        db.close()  # 确保会话关闭

7. 集成到 FastAPI 端点

现在，创建 FastAPI 应用并定义端点。我们将使用 Pydantic 模型来定义请求和响应数据。

步骤 1：创建 FastAPI 应用

app = FastAPI(title="FastAPI with SQLAlchemy Tutorial", version="1.0.0")

步骤 2：定义 Pydantic 模型（用于验证输入/输出）

class UserCreate(BaseModel):
    name: str
    email: str

class UserResponse(BaseModel):
    id: int
    name: str
    email: str

    class Config:
        orm_mode = True  # 允许从 SQLAlchemy 模型自动转换为 Pydantic 模型

步骤 3：创建 CRUD 端点

例如，一个创建用户的端点：

@app.post("/users/", response_model=UserResponse)
def create_user(user: UserCreate, db: Session = Depends(get_db)):
    """
    创建新用户。
    - `user`: 来自请求体的 UserCreate 数据。
    - `db`: 通过依赖注入获取的数据库会话。
    """
    # 将 Pydantic 模型转换为 SQLAlchemy 模型
    db_user = User(name=user.name, email=user.email)
    db.add(db_user)  # 添加到会话
    db.commit()  # 提交事务
    db.refresh(db_user)  # 刷新以获取生成的 ID 等
    return db_user  # 自动转换为 UserResponse 响应模型

类似地，可以定义读取、更新和删除端点。

8. 示例：完整的 CRUD 操作

这是一个简单的示例，展示如何在单个文件中整合所有内容。保存为 main.py 并运行 uvicorn main:app --reload 启动服务。

from fastapi import FastAPI, Depends, HTTPException
from sqlalchemy import create_engine, Column, Integer, String
from sqlalchemy.ext.declarative import declarative_base
from sqlalchemy.orm import sessionmaker, Session
from pydantic import BaseModel
from typing import List

# 数据库配置
SQLALCHEMY_DATABASE_URL = "sqlite:///./test.db"
engine = create_engine(SQLALCHEMY_DATABASE_URL, connect_args={"check_same_thread": False})
SessionLocal = sessionmaker(autocommit=False, autoflush=False, bind=engine)
Base = declarative_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)

# 创建表
Base.metadata.create_all(bind=engine)

# Pydantic 模型
class UserCreate(BaseModel):
    name: str
    email: str

class UserResponse(BaseModel):
    id: int
    name: str
    email: str
    class Config:
        orm_mode = True

# FastAPI 应用
app = FastAPI()

# 依赖函数
def get_db():
    db = SessionLocal()
    try:
        yield db
    finally:
        db.close()

# 端点
@app.post("/users/", response_model=UserResponse)
def create_user(user: UserCreate, db: Session = Depends(get_db)):
    db_user = User(name=user.name, email=user.email)
    db.add(db_user)
    db.commit()
    db.refresh(db_user)
    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 = db.query(User).offset(skip).limit(limit).all()
    return users

@app.get("/users/{user_id}", response_model=UserResponse)
def read_user(user_id: int, db: Session = Depends(get_db)):
    user = db.query(User).filter(User.id == user_id).first()
    if user is None:
        raise HTTPException(status_code=404, detail="User not found")
    return user

@app.put("/users/{user_id}", response_model=UserResponse)
def update_user(user_id: int, user: UserCreate, db: Session = Depends(get_db)):
    db_user = db.query(User).filter(User.id == user_id).first()
    if db_user is None:
        raise HTTPException(status_code=404, detail="User not found")
    db_user.name = user.name
    db_user.email = user.email
    db.commit()
    db.refresh(db_user)
    return db_user

@app.delete("/users/{user_id}")
def delete_user(user_id: int, db: Session = Depends(get_db)):
    db_user = db.query(User).filter(User.id == user_id).first()
    if db_user is None:
        raise HTTPException(status_code=404, detail="User not found")
    db.delete(db_user)
    db.commit()
    return {"detail": "User deleted"}

运行后，访问 http://localhost:8000/docs 查看自动生成的交互式 API 文档。

9. 最佳实践与常见问题

最佳实践

• 使用依赖注入：始终通过 Depends(get_db) 管理会话，避免手动打开和关闭。
• 事务管理：在端点中显式提交事务（db.commit()），并处理异常以确保回滚（可使用 try-except）。
• 模型分离：考虑将 SQLAlchemy 模型、Pydantic 模型和业务逻辑分开到不同文件中以提高可维护性。
• 错误处理：使用 FastAPI 的 HTTPException 返回适当的错误响应。
• 性能优化：对频繁查询的字段添加索引（如示例中的 index=True）。

常见问题

• 连接泄漏：确保每个请求后关闭会话；依赖函数 get_db 通过 finally 块处理。
• 循环导入：如果分模块组织，注意导入顺序以避免循环依赖。
• 数据库迁移：对于生产环境，建议使用 Alembic（SQLAlchemy 的迁移工具）来管理模式变更。
• 并发问题：SQLAlchemy 会话默认是线程不安全的；在 FastAPI（异步）中，使用 SessionLocal 创建新会话，通常没问题，但需注意会话生命周期。

10. 总结与下一步

在本教程中，你学习了如何将 SQLAlchemy 核心集成到 FastAPI 中，包括设置数据库连接、定义模型、管理会话和实现 CRUD 端点。这为你构建数据驱动的 Web 应用奠定了坚实基础。

下一步建议：

• 探索更高级的 SQLAlchemy 功能，如关系映射（一对多、多对多）。
• 集成 Alembic 进行数据库迁移。
• 学习 FastAPI 的异步特性，结合 SQLAlchemy 的异步支持（如使用 sqlalchemy.ext.asyncio）。
• 参考官方文档：FastAPI SQLAlchemy 示例。

祝你学习愉快！如果有问题，欢迎查阅社区资源或继续深入本教程的其他章节。

注意：本教程使用 SQLite 示例，实际项目中请替换为适合的数据库（如 PostgreSQL）并调整连接参数。代码示例已简化，建议在生产环境中添加更多错误处理和安全性措施。