logo

开发者热搜

FastAPI 与 Tortoise-ORM 深度集成指南

作者：很菜不狗2025.09.19 13:43浏览量：7

简介：本文详细介绍如何在 FastAPI 项目中集成 Tortoise-ORM，包括环境配置、模型定义、CRUD 操作及事务管理，助力开发者构建高效数据驱动应用。

FastAPI 集成 Tortoise-ORM 实践

在 FastAPI 项目中集成 Tortoise-ORM 已成为构建高性能数据驱动应用的热门选择。Tortoise-ORM 作为一款异步 ORM 框架，完美契合 FastAPI 的异步特性，为开发者提供了简洁高效的数据库操作方式。本文将深入探讨 FastAPI 与 Tortoise-ORM 的集成实践，涵盖环境配置、模型定义、CRUD 操作及事务管理等关键环节。

一、环境准备与基础配置

1.1 安装必要依赖

首先需要安装 FastAPI、Uvicorn（ASGI 服务器）和 Tortoise-ORM 核心包：

  1. pip install fastapi uvicorn tortoise-orm

对于 PostgreSQL 数据库，还需安装 asyncpg 驱动：

  1. pip install asyncpg

1.2 项目结构规划

推荐采用模块化结构组织项目：

  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
  11. │   └── crud/            # 数据库操作
  12. │       ├── __init__.py
  13. │       └── user.py
  14. └── requirements.txt

1.3 Tortoise-ORM 初始化配置

在 app/db.py 中配置数据库连接：

  1. from tortoise import Tortoise
  3. async def init_db():
  4.     await Tortoise.init(
  5.         db_url="postgres://user:password@localhost:5432/mydb",
  6.         modules={"models": ["app.models"]}
  7.     )
  8.     await Tortoise.generate_schemas()  # 自动生成表结构
  10. async def close_db():
  11.     await Tortoise.close_connections()

在 FastAPI 启动时初始化数据库：

  1. from fastapi import FastAPI
  2. from app.db import init_db, close_db
  4. app = FastAPI()
  6. @app.on_event("startup")
  7. async def startup_event():
  8.     await init_db()
  10. @app.on_event("shutdown")
  11. async def shutdown_event():
  12.     await close_db()

二、模型定义与关系映射

2.1 基础模型定义

在 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)
  9.     updated_at = fields.DatetimeField(auto_now=True)
  11.     def __str__(self):
  12.         return self.username

2.2 关系映射实践

定义一对多关系（用户-文章）：

  1. class Article(models.Model):
  2.     id = fields.IntField(pk=True)
  3.     title = fields.CharField(max_length=255)
  4.     content = fields.TextField()
  5.     author = fields.ForeignKeyField("models.User", related_name="articles")
  6.     created_at = fields.DatetimeField(auto_now_add=True)
  8. # 在 User 模型中添加反向引用
  9. class User(models.Model):
  10.     # ... 其他字段 ...
  11.     articles = fields.ReverseRelation["Article"]()

2.3 模型继承与多态

实现基础模型继承：

  1. class TimestampModel(models.Model):
  2.     created_at = fields.DatetimeField(auto_now_add=True)
  3.     updated_at = fields.DatetimeField(auto_now=True)
  5.     class Meta:
  6.         abstract = True  # 标记为抽象模型
  8. class Product(TimestampModel):
  9.     id = fields.IntField(pk=True)
  10.     name = fields.CharField(max_length=100)
  11.     price = fields.DecimalField(max_digits=10, decimal_places=2)

三、CRUD 操作实现

3.1 创建操作

在 app/crud/user.py 中实现用户创建：

  1. from app.models import User
  2. from app.schemas.user import UserCreate
  4. async def create_user(user_data: UserCreate):
  5.     user = await User.create(**user_data.dict())
  6.     return user

3.2 查询操作

实现多种查询方式：

  1. # 获取所有活跃用户
  2. async def get_active_users():
  3.     return await User.filter(is_active=True).all()
  5. # 分页查询
  6. async def get_users_paginated(skip: int = 0, limit: int = 10):
  7.     return await User.all().offset(skip).limit(limit)
  9. # 复杂条件查询
  10. async def get_users_by_criteria(username: str = None, email: str = None):
  11.     query = User.filter()
  12.     if username:
  13.         query = query.filter(username__icontains=username)
  14.     if email:
  15.         query = query.filter(email__icontains=email)
  16.     return await query.all()

3.3 更新与删除

实现安全的更新和删除操作：

  1. # 更新用户信息
  2. async def update_user(user_id: int, update_data: dict):
  3.     await User.filter(id=user_id).update(**update_data)
  4.     return await User.get(id=user_id)
  6. # 软删除实现
  7. async def soft_delete_user(user_id: int):
  8.     await User.filter(id=user_id).update(is_active=False)
  9.     return True
  11. # 硬删除（谨慎使用）
  12. async def hard_delete_user(user_id: int):
  13.     await User.filter(id=user_id).delete()
  14.     return True

四、事务管理与高级特性

4.1 原子事务操作

使用 atomic() 装饰器确保事务完整性：

  1. from tortoise import transactions
  3. @transactions.atomic()
  4. async def transfer_funds(from_id: int, to_id: int, amount: float):
  5.     from_user = await User.get(id=from_id)
  6.     to_user = await User.get(id=to_id)
  8.     if from_user.balance < amount:
  9.         raise ValueError("Insufficient funds")
  11.     await User.filter(id=from_id).update(balance=from_user.balance - amount)
  12.     await User.filter(id=to_id).update(balance=to_user.balance + amount)

4.2 批量操作优化

实现高效批量插入：

  1. async def bulk_create_users(user_data_list: list):
  2.     users = [User(**data) for data in user_data_list]
  3.     await User.bulk_create(users, batch_size=100)

4.3 数据库迁移管理

使用 Aerich 进行数据库迁移：

  1. 安装迁移工具：

    1. pip install aerich

  2. 初始化配置（在 settings.py 中）：

    1. AERICH_CONFIG = {
    2.  "tortoise_orm": "app.db.TORTOISE_ORM",
    3.  "app": "app",
    4.  "migrate_dir": "migrations",
    5. }

  3. 常用迁移命令：

    1. # 初始化迁移目录
    2. aerich init -t app.db.TORTOISE_ORM
    3. # 创建迁移文件
    4. aerich migrate
    5. # 执行迁移
    6. aerich upgrade

五、性能优化与最佳实践

5.1 查询优化技巧

  1. 选择必要字段：

    1. # 只查询需要的字段
    2. async def get_user_minimal(user_id: int):
    3.  return await User.get(id=user_id).values("id", "username")

  2. 预加载关联数据：

    1. # 预加载用户文章
    2. async def get_user_with_articles(user_id: int):
    3.  return await User.get(id=user_id).prefetch_related("articles")

5.2 连接池管理

配置连接池参数：

  1. await Tortoise.init(
  2.     db_url="postgres://user:password@localhost:5432/mydb",
  3.     modules={"models": ["app.models"]},
  4.     connections={
  5.         "default": {
  6.             "engine": "tortoise.backends.asyncpg",
  7.             "credentials": {
  8.                 "host": "localhost",
  9.                 "port": "5432",
  10.                 "user": "user",
  11.                 "password": "password",
  12.                 "database": "mydb",
  13.                 "minsize": 5,
  14.                 "maxsize": 20,
  15.             }
  16.         }
  17.     }
  18. )

5.3 缓存策略集成

结合 Redis 实现查询缓存：

  1. from fastapi_cache import FastAPICache
  2. from fastapi_cache.backends.redis import RedisBackend
  3. from redis import asyncio as aioredis
  5. async def init_cache():
  6.     redis = aioredis.from_url("redis://localhost")
  7.     FastAPICache.init(RedisBackend(redis), prefix="fastapi-cache")
  9. # 在路由中使用缓存
  10. @app.get("/users/{user_id}")
  11. @cache(expire=60)  # 缓存60秒
  12. async def get_user(user_id: int):
  13.     return await User.get(id=user_id)

六、完整 API 示例

6.1 用户管理 API

  1. from fastapi import APIRouter, HTTPException
  2. from app.crud.user import create_user, get_user, update_user, delete_user
  3. from app.schemas.user import UserCreate, UserUpdate, UserDisplay
  5. router = APIRouter(prefix="/users", tags=["users"])
  7. @router.post("/", response_model=UserDisplay)
  8. async def create_new_user(user: UserCreate):
  9.     db_user = await create_user(user)
  10.     return db_user
  12. @router.get("/{user_id}", response_model=UserDisplay)
  13. async def read_user(user_id: int):
  14.     db_user = await get_user(user_id)
  15.     if db_user is None:
  16.         raise HTTPException(status_code=404, detail="User not found")
  17.     return db_user
  19. @router.patch("/{user_id}", response_model=UserDisplay)
  20. async def update_user_profile(user_id: int, user: UserUpdate):
  21.     db_user = await update_user(user_id, user.dict(exclude_unset=True))
  22.     if db_user is None:
  23.         raise HTTPException(status_code=404, detail="User not found")
  24.     return db_user
  26. @router.delete("/{user_id}")
  27. async def delete_user_profile(user_id: int):
  28.     success = await delete_user(user_id)
  29.     if not success:
  30.         raise HTTPException(status_code=404, detail="User not found")
  31.     return {"message": "User deleted successfully"}

6.2 文章管理 API

  1. from fastapi import APIRouter, HTTPException
  2. from app.models import Article
  3. from app.schemas.article import ArticleCreate, ArticleDisplay
  5. router = APIRouter(prefix="/articles", tags=["articles"])
  7. @router.post("/", response_model=ArticleDisplay)
  8. async def create_article(article: ArticleCreate, current_user_id: int):
  9.     article_data = article.dict()
  10.     article_data["author_id"] = current_user_id
  11.     db_article = await Article.create(**article_data)
  12.     return db_article
  14. @router.get("/{article_id}", response_model=ArticleDisplay)
  15. async def read_article(article_id: int):
  16.     db_article = await Article.get_or_none(id=article_id)
  17.     if db_article is None:
  18.         raise HTTPException(status_code=404, detail="Article not found")
  19.     return db_article

七、常见问题解决方案

7.1 循环导入问题

解决方案：使用延迟导入或重构项目结构

  1. # 错误示例（直接导入模型）
  2. # from app.models import User  # 可能导致循环导入
  4. # 正确做法（在函数内导入）
  5. async def get_user_articles(user_id: int):
  6.     from app.models import User, Article  # 函数内导入
  7.     user = await User.get(id=user_id)
  8.     return await user.articles.all()

7.2 事务回滚处理

  1. from tortoise.exceptions import OperationalError
  3. async def safe_transfer(from_id: int, to_id: int, amount: float):
  4.     try:
  5.         async with transactions.atomic():
  6.             await transfer_funds(from_id, to_id, amount)
  7.     except OperationalError as e:
  8.         print(f"Transaction failed: {str(e)}")
  9.         raise HTTPException(status_code=500, detail="Transaction failed")

7.3 性能监控

集成 Prometheus 监控：

  1. from prometheus_fastapi_instrumentator import Instrumentator
  3. app = FastAPI()
  4. Instrumentator().instrument(app).expose(app)

八、总结与展望

FastAPI 与 Tortoise-ORM 的集成提供了构建现代 Web 应用的强大工具链。通过异步架构设计，开发者能够充分利用现代硬件资源，构建高并发、低延迟的数据服务。未来发展方向包括：

  1. 更完善的分布式事务支持
  2. 与 GraphQL 的深度集成
  3. 自动化代码生成工具的增强
  4. 多数据库后端的无缝切换

建议开发者持续关注 Tortoise-ORM 的更新日志，及时应用新特性优化应用性能。同时，建议建立完善的测试体系，确保数据库操作的安全性和可靠性。

相关文章推荐

发表评论

最热文章

关于作者

  • 被阅读数
  • 被赞数
  • 被收藏数
活动
咨询