---

FastAPI 整合 ORM 教程：拥抱 SQLAlchemy 与 Tortoise-ORM

FastAPI 以其高性能、易用性以及对现代 Python 特性（如类型提示和异步）的出色支持，迅速成为构建 API 的热门框架。然而，任何严肃的 Web 应用程序都需要与数据库进行交互。对象关系映射（ORM）工具通过将数据库表映射到 Python 对象，极大地简化了数据库操作，使开发者能够使用更 Pythonic 的方式来处理数据，而不是编写原始 SQL 语句。

本文将深入探讨如何在 FastAPI 项目中整合两种流行的 Python ORM 库：功能强大且成熟的 SQLAlchemy（特别是其异步支持）以及专为 asyncio 设计、API 简洁的 Tortoise-ORM。我们将涵盖从环境设置、模型定义、数据库初始化到实现 CRUD（创建、读取、更新、删除）操作的完整流程，并比较两者的优劣，帮助你根据项目需求做出选择。

为什么在 FastAPI 中使用 ORM？

1. 开发效率：ORM 让你使用 Python 代码操作数据库，避免编写和调试复杂的 SQL 语句，显著提高开发速度。
2. 代码可读性与可维护性：面向对象的数据库操作使代码更易于理解和维护。模型定义清晰地描述了数据结构。
3. 数据库无关性：大多数 ORM 支持多种数据库后端。理论上，更换数据库只需修改配置，而无需重写大量数据访问代码（尽管实践中可能需要注意特定数据库的特性）。
4. 安全性：ORM 通常能自动处理 SQL 注入等安全问题，通过参数化查询保护你的应用。
5. 与 FastAPI 的契合度：
   • 类型提示：FastAPI 强依赖 Pydantic 进行数据验证和序列化，而现代 ORM（如 SQLAlchemy 2.0+ 和 Tortoise-ORM）与类型提示和 Pydantic 能够良好集成，实现端到端的数据类型安全。
   • 异步支持：FastAPI 是一个 ASGI 框架，天然支持异步操作。使用支持异步的 ORM（如 SQLAlchemy 的 asyncio 扩展或 Tortoise-ORM）可以充分利用 FastAPI 的异步能力，构建非阻塞、高并发的数据库密集型应用。

准备工作

在开始之前，确保你已具备：

• Python 3.7+ 环境。
• 对 FastAPI 和 Pydantic 的基本了解。
• 对 SQL 和关系型数据库的基本概念有所了解。
• 安装了你选择的数据库（例如 PostgreSQL, MySQL, SQLite）。
• 使用虚拟环境（强烈推荐）：
  bash
python -m venv venv
source venv/bin/activate # Linux/macOS
# venv\Scripts\activate # Windows

接下来，安装 FastAPI 和 Uvicorn (ASGI 服务器)：

bash
pip install fastapi uvicorn[standard]

我们将分别介绍 SQLAlchemy 和 Tortoise-ORM，你可以根据需要安装对应的库。

第一部分：FastAPI 与 SQLAlchemy (Asyncio 模式)

SQLAlchemy 是 Python 生态中最成熟、功能最全面的 ORM 之一。从 1.4 版本开始，它引入了对 asyncio 的原生支持，并在 2.0 版本中将其作为核心特性，非常适合与 FastAPI 配合使用。

1. 安装 SQLAlchemy 及异步驱动

你需要安装 SQLAlchemy 核心库以及针对你的数据库的异步驱动程序。例如，对于 PostgreSQL，推荐使用 asyncpg；对于 MySQL，可以使用 aiomysql。

“`bash

安装 SQLAlchemy 2.0+ (会自动包含 asyncio 支持)

pip install “sqlalchemy[asyncio]”

安装 PostgreSQL 异步驱动

pip install asyncpg

或者安装 MySQL 异步驱动

pip install aiomysql

“`

2. 配置数据库连接

在项目中创建一个配置文件（如 config.py）或使用环境变量来管理数据库连接 URL。

“`python

config.py

DATABASE_URL = “postgresql+asyncpg://user:password@host:port/database_name”

例如: DATABASE_URL = “postgresql+asyncpg://postgres:secret@localhost:5432/fastapi_orm_db”

对于 SQLite (异步): DATABASE_URL = “sqlite+aiosqlite:///./test.db”

“`

3. 设置 SQLAlchemy 引擎和会话

创建一个 database.py 文件来管理数据库引擎和会话。

“`python

database.py

from sqlalchemy.ext.asyncio import create_async_engine, AsyncSession, async_sessionmaker
from sqlalchemy.orm import declarative_base
from config import DATABASE_URL

创建异步引擎

echo=True 会打印执行的 SQL 语句，便于调试

engine = create_async_engine(DATABASE_URL, echo=True)

创建异步会话工厂

expire_on_commit=False 防止在提交后访问对象时需要重新查询

AsyncSessionFactory = async_sessionmaker(
bind=engine,
class_=AsyncSession,
expire_on_commit=False
)

创建 Declarative Base

Base = declarative_base()

依赖注入函数：获取数据库会话

async def get_db() -> AsyncSession:
async with AsyncSessionFactory() as session:
try:
yield session
finally:
await session.close()
“`

关键点解释：

• create_async_engine: 创建一个异步数据库引擎。
• async_sessionmaker: 创建一个工厂函数，用于生成 AsyncSession 实例。这是进行数据库操作的主要接口。
• declarative_base: 所有 ORM 模型都需要继承这个基类。
• get_db: 这是一个异步生成器函数，设计为 FastAPI 的依赖项。它负责创建会话、在请求处理函数中使用会话（通过 yield），并在请求结束后（无论成功或失败）确保会话被关闭。这是管理数据库会话的标准模式。

4. 定义 ORM 模型

在 models.py 文件中定义你的数据模型。

“`python

models.py

from sqlalchemy import Column, Integer, String, DateTime
from sqlalchemy.sql import func
from database import Base

class Item(Base):
tablename = “items”

id = Column(Integer, primary_key=True, index=True)
name = Column(String(50), unique=True, index=True, nullable=False)
description = Column(String(200), nullable=True)
created_at = Column(DateTime(timezone=True), server_default=func.now())
updated_at = Column(DateTime(timezone=True), onupdate=func.now())

def __repr__(self):
    return f"<Item(name='{self.name}', description='{self.description}')>"

“`

这里我们定义了一个 Item 模型，对应数据库中的 items 表。我们使用了 SQLAlchemy 的 Column 来定义字段及其类型和约束。

5. 定义 Pydantic Schema

为了进行数据验证和序列化，我们需要定义 Pydantic 模型（通常放在 schemas.py）。将 API 的数据结构与数据库模型分离是一种良好的实践。

“`python

schemas.py

from pydantic import BaseModel
from datetime import datetime
from typing import Optional

用于创建 Item 的基础模型 (输入)

class ItemBase(BaseModel):
name: str
description: Optional[str] = None

用于创建 Item 的模型 (继承自 Base, API 输入)

class ItemCreate(ItemBase):
pass

用于更新 Item 的模型 (所有字段可选)

class ItemUpdate(BaseModel):
name: Optional[str] = None
description: Optional[str] = None

用于从数据库读取 Item 的模型 (继承自 Base, 包含 id 和时间戳, API 输出)

class ItemRead(ItemBase):
id: int
created_at: datetime
updated_at: Optional[datetime] = None # onupdate 可能首次为 None

class Config:
    orm_mode = True # Pydantic V1
    # from_attributes = True # Pydantic V2

“`

orm_mode = True (或 Pydantic V2 的 from_attributes = True) 使得 Pydantic 模型能够直接从 ORM 对象（如我们查询到的 Item 实例）读取数据。

6. 实现 CRUD 操作

现在，在 main.py 中创建 FastAPI 应用并定义 API 端点。

“`python

main.py

from fastapi import FastAPI, Depends, HTTPException, status
from sqlalchemy.ext.asyncio import AsyncSession
from sqlalchemy.future import select
from typing import List

import models
import schemas
from database import engine, Base, get_db

app = FastAPI()

应用启动时创建数据库表 (生产环境建议使用 Alembic 进行迁移管理)

@app.on_event(“startup”)
async def startup_event():
async with engine.begin() as conn:
# await conn.run_sync(Base.metadata.drop_all) # 可选：开发时清空
await conn.run_sync(Base.metadata.create_all)

创建 Item

@app.post(“/items/”, response_model=schemas.ItemRead, status_code=status.HTTP_201_CREATED)
async def create_item(item: schemas.ItemCreate, db: AsyncSession = Depends(get_db)):
db_item = models.Item(**item.dict())
db.add(db_item)
await db.commit()
await db.refresh(db_item) # 刷新以获取数据库生成的值（如 id, created_at）
return db_item

读取单个 Item

@app.get(“/items/{item_id}”, response_model=schemas.ItemRead)
async def read_item(item_id: int, db: AsyncSession = Depends(get_db)):
# SQLAlchemy 2.0 推荐使用 session.get() 获取主键对象
db_item = await db.get(models.Item, item_id)
if db_item is None:
raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail=”Item not found”)
return db_item

读取多个 Items (带分页)

@app.get(“/items/”, response_model=List[schemas.ItemRead])
async def read_items(skip: int = 0, limit: int = 100, db: AsyncSession = Depends(get_db)):
# SQLAlchemy 2.0 风格的查询
stmt = select(models.Item).offset(skip).limit(limit)
result = await db.execute(stmt)
items = result.scalars().all() # 获取所有 Item 对象
return items

更新 Item

@app.put(“/items/{item_id}”, response_model=schemas.ItemRead)
async def update_item(item_id: int, item: schemas.ItemUpdate, db: AsyncSession = Depends(get_db)):
db_item = await db.get(models.Item, item_id)
if db_item is None:
raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail=”Item not found”)

# 获取更新数据，排除未设置的字段
update_data = item.dict(exclude_unset=True)
for key, value in update_data.items():
    setattr(db_item, key, value)

db.add(db_item) # 标记对象为 'dirty'
await db.commit()
await db.refresh(db_item)
return db_item

删除 Item

@app.delete(“/items/{item_id}”, status_code=status.HTTP_204_NO_CONTENT)
async def delete_item(item_id: int, db: AsyncSession = Depends(get_db)):
db_item = await db.get(models.Item, item_id)
if db_item is None:
raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail=”Item not found”)

await db.delete(db_item)
await db.commit()
return None # 对于 204 状态码，FastAPI 不应返回内容

“`

运行应用:

bash
uvicorn main:app --reload

现在你可以通过 API 文档 (通常是 /docs) 或 curl 等工具测试这些端点。

注意: Base.metadata.create_all 适用于开发或简单场景。在生产环境中，强烈建议使用数据库迁移工具如 Alembic 来管理数据库模式的变更。

第二部分：FastAPI 与 Tortoise-ORM

Tortoise-ORM 是一个易于使用、受 Django ORM 启发的 asyncio ORM。它完全基于异步设计，API 简洁直观，并且与 Pydantic 有着非常好的原生集成。

1. 安装 Tortoise-ORM 及驱动

“`bash

安装 Tortoise-ORM

pip install tortoise-orm

安装数据库驱动 (以 asyncpg 为例)

pip install asyncpg

或者其他驱动，如 aiosqlite, aiomysql

“`

2. 配置 Tortoise-ORM

Tortoise-ORM 通常通过一个字典进行配置。在 main.py 或单独的配置文件中定义配置。

“`python

main.py (或者 config.py)

Tortoise ORM 配置

TORTOISE_ORM = {
“connections”: {
# “default” 是连接名，可以自定义
“default”: “postgres://user:password@host:port/database_name”
# 例如: “default”: “postgres://postgres:secret@localhost:5432/fastapi_tortoise_db”
# 对于 SQLite: “default”: “sqlite://./tortoise_test.db”
},
“apps”: {
“models”: { # “models” 是 app 名，可以自定义
“models”: [“models_tortoise”, “aerich.models”], # 指定模型文件路径，aerich.models 用于迁移
“default_connection”: “default”, # 此 app 使用哪个连接
}
},
“use_tz”: False, # 是否使用时区，建议为 True
“timezone”: “UTC” # 如果 use_tz=True
}
“`

关键点解释:

• connections: 定义数据库连接 URL。
• apps: Tortoise 采用类似 Django 的 “app” 概念来组织模型。你需要指定包含模型定义的 Python 模块路径（例如 models_tortoise.py 就写 "models_tortoise"）。aerich.models 是 Tortoise 官方迁移工具 Aerich 所需的。

3. 定义 ORM 模型 (Tortoise 风格)

创建 models_tortoise.py 文件来定义模型。

“`python

models_tortoise.py

from tortoise import fields, models
from tortoise.contrib.pydantic import pydantic_model_creator

class Item(models.Model):
id = fields.IntField(pk=True)
name = fields.CharField(max_length=50, unique=True, index=True)
description = fields.TextField(null=True)
created_at = fields.DatetimeField(auto_now_add=True)
updated_at = fields.DatetimeField(auto_now=True)

class Meta:
    table = "items_tortoise" # 可以指定表名，默认会根据 app 和类名生成

def __str__(self):
    return self.name

自动创建 Pydantic 模型 (非常方便!)

Item_Pydantic = pydantic_model_creator(Item, name=”Item”) # 用于读取 (输出)
ItemIn_Pydantic = pydantic_model_creator(Item, name=”ItemIn”, exclude_readonly=True) # 用于创建/更新 (输入)
“`

Tortoise 的模型定义与 Django 非常相似。特别注意 pydantic_model_creator，它可以根据 Tortoise 模型自动生成 Pydantic schema，大大减少了重复代码。exclude_readonly=True 会排除像 id, created_at, updated_at 这样通常由数据库或 ORM 自动管理的字段，适合作为创建和更新操作的输入模型。

4. 初始化 Tortoise-ORM 和数据库

在 FastAPI 应用启动和关闭时，需要初始化和关闭 Tortoise 连接。可以使用 register_tortoise 辅助函数简化这个过程。

“`python

main_tortoise.py

from fastapi import FastAPI, Depends, HTTPException, status
from tortoise.contrib.fastapi import register_tortoise
from tortoise.exceptions import DoesNotExist, IntegrityError
from typing import List

从 models_tortoise 导入模型和 Pydantic Schema

from models_tortoise import Item, Item_Pydantic, ItemIn_Pydantic
from config import TORTOISE_ORM # 假设配置在 config.py

app = FastAPI()

使用 register_tortoise 注册 Tortoise ORM

它会在应用启动时初始化 Tortoise，并在关闭时断开连接

generate_schemas=True 会在启动时创建表 (仅建议开发时使用)

register_tortoise(
app,
config=TORTOISE_ORM,
generate_schemas=True, # 生产环境应设为 False，并使用 Aerich 管理迁移
add_exception_handlers=True, # 自动添加处理 DoesNotExist 等异常的处理器
)

注意：Tortoise 不需要像 SQLAlchemy 那样的显式 Session 依赖注入。

Tortoise 的操作是直接在模型类或实例上进行的，它内部管理连接和事务。

创建 Item

@app.post(“/items/”, response_model=Item_Pydantic, status_code=status.HTTP_201_CREATED)
async def create_item(item: ItemIn_Pydantic):
try:
# Tortoise 的 create 方法直接接受 Pydantic 模型的 dict
db_item = await Item.create(**item.dict(exclude_unset=True))
return await Item_Pydantic.from_tortoise_orm(db_item)
except IntegrityError: # 处理唯一约束冲突等
raise HTTPException(
status_code=status.HTTP_400_BAD_REQUEST,
detail=f”Item with name ‘{item.name}’ already exists.”,
)

读取单个 Item

@app.get(“/items/{item_id}”, response_model=Item_Pydantic)
async def read_item(item_id: int):
try:
db_item = await Item.get(id=item_id)
return await Item_Pydantic.from_tortoise_orm(db_item)
except DoesNotExist:
raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail=”Item not found”)

读取多个 Items

@app.get(“/items/”, response_model=List[Item_Pydantic])
async def read_items(skip: int = 0, limit: int = 100):
items = await Item.all().offset(skip).limit(limit)
return await Item_Pydantic.from_queryset(items) # 从查询集转换

更新 Item

@app.put(“/items/{item_id}”, response_model=Item_Pydantic)
async def update_item(item_id: int, item: ItemIn_Pydantic):
try:
# 更新指定字段，返回影响的行数
await Item.filter(id=item_id).update(**item.dict(exclude_unset=True))
# 更新后需要重新获取对象以返回最新数据
updated_item = await Item.get(id=item_id)
return await Item_Pydantic.from_tortoise_orm(updated_item)
except DoesNotExist:
raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail=”Item not found”)
except IntegrityError: # 处理可能的唯一约束冲突
raise HTTPException(
status_code=status.HTTP_400_BAD_REQUEST,
detail=f”Cannot update item, potential conflict (e.g., name already exists).”,
)

删除 Item

@app.delete(“/items/{item_id}”, status_code=status.HTTP_204_NO_CONTENT)
async def delete_item(item_id: int):
try:
deleted_count = await Item.filter(id=item_id).delete()
if deleted_count == 0:
raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail=”Item not found”)
except DoesNotExist: # 理论上 filter().delete() 不会抛这个，但以防万一
raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail=”Item not found”)
return None
“`

运行应用:

bash
uvicorn main_tortoise:app --reload

关键点:

• register_tortoise: 极大地简化了 Tortoise 的初始化和关闭流程。
• 无显式会话管理: Tortoise ORM 在后台处理连接和事务，API 调用更直接（例如 Item.create(), Item.get(), Item.filter().update()）。
• Pydantic 集成: pydantic_model_creator 和 from_tortoise_orm/from_queryset 方法使得模型和 Schema 之间的转换非常流畅。
• 异常处理: DoesNotExist 和 IntegrityError 是常见的需要处理的 Tortoise 异常。add_exception_handlers=True 可以帮助自动转换 DoesNotExist 为 404 响应。
• 数据库迁移: Tortoise 推荐使用 Aerich 进行数据库迁移。你需要安装它 (pip install aerich) 并按照其文档进行初始化和管理迁移。

SQLAlchemy vs Tortoise-ORM：如何选择？

特性	SQLAlchemy (Asyncio)	Tortoise-ORM
成熟度	非常高，久经考验	相对年轻，但发展迅速
功能	极其丰富，支持复杂查询、Core API、多种数据库特性	核心 ORM 功能完善，API 更简洁，侧重常用功能
异步支持	1.4+ 原生支持，2.0 成为核心	从一开始就为 asyncio 设计
API 风格	灵活，功能强大，但有时可能稍显冗余（尤其会话管理）	简洁，类似 Django ORM，学习曲线平缓
Pydantic集成	需要手动定义 Pydantic 模型，使用 orm_mode/from_attributes	内建 pydantic_model_creator，集成度高
会话管理	需要显式管理 AsyncSession（通常通过依赖注入）	隐式管理，API 调用更直接
社区与生态	非常庞大，文档、教程、第三方库丰富	较小，但活跃，增长中
迁移工具	Alembic (事实标准)	Aerich (官方推荐)
适用场景	复杂应用、需要精细控制 SQL、已有 SQLAlchemy 经验、同步/异步混合	新的 Async-first 项目、追求简洁 API、Django 开发者迁移

选择建议:

• 如果你需要最强大的功能、对 SQL 的精细控制，或者你的团队已经熟悉 SQLAlchemy，那么选择 SQLAlchemy 的异步模式是明智的。虽然需要多写一点模板代码（如会话管理），但其灵活性和成熟度是巨大优势。
• 如果你正在开始一个全新的、完全基于 asyncio 的项目，并且偏好简洁、类似 Django 的 API 和与 Pydantic 的无缝集成，Tortoise-ORM 是一个非常有吸引力的选择。它的学习曲线更平缓，能让你更快地投入业务逻辑开发。

最佳实践与注意事项

1. 异步优先: 既然选择了 FastAPI，就应该尽可能使用异步数据库驱动和异步 ORM 操作，避免阻塞事件循环。
2. 依赖注入 (SQLAlchemy): 正确使用 FastAPI 的 Depends 来管理 SQLAlchemy 的 AsyncSession 生命周期至关重要。
3. 分离 Schema: 始终将用于 API 输入/输出的 Pydantic Schema 与 ORM 模型分开定义。这提供了更好的抽象和灵活性。
4. 数据库迁移: 绝不要在生产环境中使用 Base.metadata.create_all 或 generate_schemas=True。务必使用 Alembic (SQLAlchemy) 或 Aerich (Tortoise-ORM) 来管理数据库模式的演进。
5. 错误处理: 妥善处理数据库操作可能抛出的异常（如 DoesNotExist, IntegrityError），并将它们转换为合适的 HTTP 响应。
6. 配置管理: 使用环境变量或配置文件（如 Pydantic 的 BaseSettings）来管理数据库连接信息和其他配置，不要硬编码在代码中。
7. 测试: 编写集成测试时，需要设置测试数据库，并在测试运行前后正确地创建/销毁表或回滚事务。
8. 性能: 对于性能敏感的查询，了解 ORM 生成的 SQL 语句（SQLAlchemy 的 echo=True 或 Tortoise 的日志配置），并考虑使用 ORM 提供的优化手段（如 select_related, prefetch_related）。

总结

将 ORM 整合到 FastAPI 应用中是构建健壮、可维护的数据驱动 API 的关键一步。SQLAlchemy 以其无与伦比的功能和成熟度提供了强大的异步支持，而 Tortoise-ORM 则以其简洁的 API、出色的 Pydantic 集成和纯粹的异步设计吸引了众多开发者。

理解这两种 ORM 的设置、模型定义、CRUD 操作模式以及它们与 FastAPI 的交互方式，将使你能够根据项目需求做出最佳技术选型。无论你选择哪一个，遵循异步编程的最佳实践、使用数据库迁移工具并妥善管理数据库连接/会话，都将帮助你构建出高性能、高效率的 FastAPI 应用。

希望这篇详细的教程能帮助你成功地在 FastAPI 项目中使用 SQLAlchemy 或 Tortoise-ORM！

---