FastAPI与Tortoise-ORM集成全攻略

2025年10月17日 互联网

FastAPI集成Tortoise-ORM实践

引言

在Python生态中，FastAPI凭借其高性能、易用性和现代特性（如类型注解、异步支持）迅速成为构建API服务的热门选择。而Tortoise-ORM作为一款专为异步框架设计的ORM工具，完美契合FastAPI的异步特性，提供了直观的数据库操作接口。本文将深入探讨如何在FastAPI项目中集成Tortoise-ORM，从基础配置到高级用法，为开发者提供一套完整的实践指南。

环境准备与基础配置

安装依赖

首先，确保你的开发环境已安装Python 3.7+。接着，通过pip安装FastAPI、Uvicorn（ASGI服务器）和Tortoise-ORM：

  1. pip install fastapi uvicorn tortoise-orm

项目结构规划

一个良好的项目结构有助于代码管理和维护。推荐采用以下结构：

  1. project/
  2. ├── app/
  3. │   ├── __init__.py
  4. │   ├── main.py          # FastAPI应用入口
  5. │   ├── models/          # 数据库模型定义
  6. │   │   ├── __init__.py
  7. │   │   └── user.py      # 用户模型示例
  8. │   ├── schemas/         # Pydantic模型，用于请求/响应验证
  9. │   │   ├── __init__.py
  10. │   │   └── user.py      # 用户Schema示例
  11. │   ├── crud/            # 数据库操作逻辑
  12. │   │   ├── __init__.py
  13. │   │   └── user.py      # 用户CRUD操作
  14. │   └── config.py        # 配置文件，如数据库连接
  15. └── requirements.txt

配置Tortoise-ORM

在config.py中定义数据库连接配置：

  1. from tortoise.contrib.fastapi import register_tortoise
  3. DB_CONFIG = {
  4.     "connections": {
  5.         "default": {
  6.             "engine": "tortoise.backends.asyncpg",
  7.             "credentials": {
  8.                 "host": "localhost",
  9.                 "port": "5432",
  10.                 "user": "postgres",
  11.                 "password": "yourpassword",
  12.                 "database": "mydatabase",
  13.             },
  14.         },
  15.     },
  16.     "apps": {
  17.         "models": {
  18.             "models": ["app.models"],
  19.             "default_connection": "default",
  20.         },
  21.     },
  22.     "use_tz": True,
  23.     "timezone": "UTC",
  24. }

模型定义与数据库迁移

定义模型

在app/models/user.py中定义用户模型：

  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.     def __str__(self):
  11.         return self.username

数据库迁移

Tortoise-ORM不直接提供迁移工具，但可以结合Aerich（专为Tortoise设计的迁移工具）使用。首先安装Aerich：

  1. pip install aerich

初始化Aerich配置（通常在项目根目录创建aerich.ini和aerich目录）：

  1. [aerich]
  2. tortoise_orm = app.config.DB_CONFIG
  3. location = ./migrations

初始化数据库并生成迁移：

  1. aerich init -t app.config.DB_CONFIG
  2. aerich init-db
  3. # 修改模型后
  4. aerich migrate
  5. aerich upgrade

FastAPI应用集成

注册Tortoise-ORM

在app/main.py中注册Tortoise-ORM到FastAPI应用：

  1. from fastapi import FastAPI
  2. from app.config import DB_CONFIG
  3. from tortoise.contrib.fastapi import register_tortoise
  5. app = FastAPI()
  7. register_tortoise(
  8.     app,
  9.     config=DB_CONFIG,
  10.     generate_schemas=True,  # 自动生成数据库表结构
  11.     add_exception_handlers=True,
  12. )

创建CRUD操作

在app/crud/user.py中实现用户CRUD操作：

  1. from app.models import User
  2. from typing import Optional
  4. async def create_user(username: str, email: str) -> User:
  5.     user = User(username=username, email=email)
  6.     await user.save()
  7.     return user
  9. async def get_user_by_id(user_id: int) -> Optional[User]:
  10.     return await User.get_or_none(id=user_id)
  12. async def get_users() -> list[User]:
  13.     return await User.all().offset(0).limit(100)
  15. async def update_user(user_id: int, **kwargs) -> Optional[User]:
  16.     await User.filter(id=user_id).update(**kwargs)
  17.     return await get_user_by_id(user_id)
  19. async def delete_user(user_id: int) -> bool:
  20.     affected = await User.filter(id=user_id).delete()
  21.     return affected > 0

定义API路由

在app/main.py或单独的路由文件中定义API端点：

  1. from fastapi import APIRouter, HTTPException
  2. from app.crud.user import create_user, get_user_by_id, get_users, update_user, delete_user
  3. from app.schemas.user import UserCreate, UserUpdate, User
  5. router = APIRouter()
  7. @router.post("/users/", response_model=User)
  8. async def create_user_endpoint(user_in: UserCreate):
  9.     user = await create_user(username=user_in.username, email=user_in.email)
  10.     return user
  12. @router.get("/users/{user_id}", response_model=User)
  13. async def read_user_endpoint(user_id: int):
  14.     user = await get_user_by_id(user_id)
  15.     if user is None:
  16.         raise HTTPException(status_code=404, detail="User not found")
  17.     return user
  19. # 其他端点...

高级特性与最佳实践

事务处理

Tortoise-ORM支持事务，确保数据一致性：

  1. from tortoise import transactions
  3. async def transfer_funds(from_id: int, to_id: int, amount: float):
  4.     async with transactions.in_transaction() as conn:
  5.         # 假设有Account模型
  6.         from_account = await Account.get(id=from_id).using_db(conn)
  7.         to_account = await Account.get(id=to_id).using_db(conn)
  9.         if from_account.balance < amount:
  10.             raise ValueError("Insufficient funds")
  12.         from_account.balance -= amount
  13.         to_account.balance += amount
  15.         await from_account.save(using_db=conn)
  16.         await to_account.save(using_db=conn)

性能优化

  • 批量操作：使用bulk_create、bulk_update提高批量操作效率。
  • 索引优化：在频繁查询的字段上添加索引。
  • 连接池配置：根据应用负载调整数据库连接池大小。

测试策略

  • 单元测试：使用pytest和tortoise-orm.test模块进行模型测试。
  • 集成测试：使用TestClient模拟API请求，验证端到端功能。
  • 数据库隔离：测试时使用独立数据库或事务回滚，避免数据污染。

结论

FastAPI与Tortoise-ORM的集成，为开发者提供了一个高效、灵活且易于维护的API开发解决方案。通过本文的实践指南，你不仅能够快速上手基础配置和CRUD操作，还能深入理解事务处理、性能优化等高级特性。随着项目的不断演进，持续探索和优化集成方案，将助力你构建出更加健壮、高效的Web应用。

最新文章