FastAPI 集成 Tortoise-ORM 实践

2025年10月17日 互联网

FastAPI 集成 Tortoise-ORM 实践

引言

在构建现代 Web 应用时，API 的开发效率与数据持久化能力是关键。FastAPI 凭借其高性能和异步支持成为 Python 生态中的热门选择，而 Tortoise-ORM 作为一款异步 ORM 工具，能够无缝适配 FastAPI 的异步特性。本文将详细介绍如何在 FastAPI 项目中集成 Tortoise-ORM，从基础配置到高级功能，帮助开发者快速上手并优化实践。

一、环境准备与安装

1. 创建 FastAPI 项目

首先，确保已安装 Python 3.8+ 环境，并使用以下命令创建 FastAPI 项目：

  1. mkdir fastapi_tortoise_demo && cd fastapi_tortoise_demo
  2. python -m venv venv
  3. source venv/bin/activate  # Linux/macOS
  4. # 或 venv\Scripts\activate (Windows)
  5. pip install fastapi uvicorn

2. 安装 Tortoise-ORM

Tortoise-ORM 支持异步操作，需额外安装：

  1. pip install tortoise-orm[asyncpg]  # 推荐使用 asyncpg 作为 PostgreSQL 驱动

若使用 MySQL，可替换为 tortoise-orm[aiomysql]。

3. 配置数据库连接

在项目根目录创建 config.py，定义数据库连接参数：

  1. from pydantic import BaseSettings
  3. class Settings(BaseSettings):
  4.     DATABASE_URL: str = "postgres://user:password@localhost:5432/mydb"
  5.     class Config:
  6.         env_file = ".env"
  8. settings = Settings()

通过环境变量管理敏感信息，避免硬编码。

二、Tortoise-ORM 集成 FastAPI

1. 初始化 Tortoise-ORM

在 main.py 中初始化 Tortoise，并挂载到 FastAPI 应用：

  1. from fastapi import FastAPI
  2. from tortoise.contrib.fastapi import register_tortoise
  3. from config import settings
  5. app = FastAPI()
  7. register_tortoise(
  8.     app,
  9.     db_url=settings.DATABASE_URL,
  10.     modules={"models": ["app.models"]},  # 指定模型所在模块
  11.     generate_schemas=True,  # 自动生成数据库表
  12.     add_exception_handlers=True,  # 添加 ORM 异常处理
  13. )

2. 定义数据模型

在 app/models.py 中定义模型，继承自 tortoise.models.Model：

  1. from tortoise import fields, models
  3. class User(models.Model):
  4.     id = fields.IntField(pk=True)
  5.     username = fields.CharField(max_length=50, unique=True)
  6.     email = fields.CharField(max_length=255, unique=True)
  7.     is_active = fields.BooleanField(default=True)
  8.     created_at = fields.DatetimeField(auto_now_add=True)
  10.     class PydanticMeta:
  11.         computed = ["created_at"]  # 避免序列化时返回自动字段
  13.     def __str__(self):
  14.         return self.username

通过 PydanticMeta 控制序列化行为，避免泄露敏感字段。

三、CRUD 操作实现

1. 创建记录

使用 User.create() 方法异步创建用户：

  1. from fastapi import APIRouter
  2. from app.models import User
  4. router = APIRouter()
  6. @router.post("/users/")
  7. async def create_user(username: str, email: str):
  8.     user = await User.create(username=username, email=email)
  9.     return {"id": user.id, "username": user.username}

2. 查询记录

  • 查询单个用户：

    1. @router.get("/users/{user_id}/")
    2. async def get_user(user_id: int):
    3.     user = await User.get(id=user_id)
    4.     if not user:
    5.         raise HTTPException(status_code=404, detail="User not found")
    6.     return user

  • 查询所有用户：

    1. @router.get("/users/")
    2. async def list_users():
    3.     return await User.all().values("id", "username", "email")

3. 更新与删除

  • 更新用户：

    1. @router.patch("/users/{user_id}/")
    2. async def update_user(user_id: int, username: str = None, email: str = None):
    3.     await User.filter(id=user_id).update(username=username, email=email)
    4.     return {"message": "User updated"}

  • 删除用户：

    1. @router.delete("/users/{user_id}/")
    2. async def delete_user(user_id: int):
    3.     await User.filter(id=user_id).delete()
    4.     return {"message": "User deleted"}

四、高级功能实践

1. 事务管理

使用 Tortoise.transaction() 确保数据一致性：

  1. from tortoise import transactions
  3. @router.post("/transfer/")
  4. async def transfer_funds(from_id: int, to_id: int, amount: float):
  5.     async with transactions.in_transaction() as conn:
  6.         from_user = await User.get(id=from_id).using_db(conn)
  7.         to_user = await User.get(id=to_id).using_db(conn)
  9.         if from_user.balance < amount:
  10.             raise HTTPException(status_code=400, detail="Insufficient funds")
  12.         from_user.balance -= amount
  13.         to_user.balance += amount
  14.         await from_user.save(using_db=conn)
  15.         await to_user.save(using_db=conn)
  17.     return {"message": "Transfer successful"}

2. 复杂查询

  • 分页查询：

    1. @router.get("/users/page/{page}/")
    2. async def paginate_users(page: int = 1, page_size: int = 10):
    3.     skip = (page - 1) * page_size
    4.     users = await User.all().offset(skip).limit(page_size)
    5.     total = await User.all().count()
    6.     return {"data": users, "total": total}

  • 关联查询：
    假设存在 Order 模型与 User 一对多关联：

    1. class Order(models.Model):
    2.     user = fields.ForeignKeyField("models.User", related_name="orders")
    3.     amount = fields.FloatField()
    4.     # ...
    6. @router.get("/users/{user_id}/orders/")
    7. async def get_user_orders(user_id: int):
    8.     orders = await User.get(id=user_id).prefetch_related("orders")
    9.     return orders.orders

五、性能优化建议

  1. 连接池配置：
    在数据库 URL 中添加连接池参数，例如：

    1. postgres://user:password@localhost:5432/mydb?max_connections=20

  2. 选择性加载字段：
    查询时仅加载必要字段，减少数据传输：

    1. await User.all().values("id", "username")

  3. 索引优化：
    在频繁查询的字段上添加索引：

    1. class User(models.Model):
    2.     email = fields.CharField(max_length=255, unique=True, index=True)

  4. 批量操作：
    使用 bulk_create 批量插入数据：

    1. users = [User(username=f"user{i}", email=f"user{i}@example.com") for i in range(100)]
    2. await User.bulk_create(users, batch_size=50)

六、常见问题与解决方案

  1. 模型未同步：
    确保 generate_schemas=True，并检查数据库权限。

  2. 异步锁冲突：
    避免在事务中执行耗时操作，如外部 API 调用。

  3. N+1 查询问题：
    使用 prefetch_related 预加载关联数据。

总结

通过本文的实践指南，开发者可以高效地在 FastAPI 中集成 Tortoise-ORM，实现类型安全的数据库操作。从基础配置到高级事务管理，Tortoise-ORM 的异步特性与 FastAPI 完美契合，为构建高性能 Web API 提供了强大支持。建议结合实际项目需求，进一步探索 Tortoise-ORM 的插件生态（如 tortoise-migrations 数据库迁移工具），持续提升开发效率。

最新文章