FastAPI项目集成Tortoise-ORM:异步原生ORM的工程实践指南
2026/9/23 2:06:23 网站建设 项目流程

1. 为什么FastAPI项目里我会选Tortoise-ORM

先说结论:如果你正在用FastAPI写纯REST接口,又不想被迫在异步框架里写同步数据库代码,Tortoise-ORM是目前最省心的方案之一。

我第一次在FastAPI里用SQLAlchemy时,遇到的第一个坑就是同步Session和异步事件循环之间的纠缠。虽然SQLAlchemy 2.0推出了AsyncSession,但用起来总有一种"为了异步而异步"的拧巴感——模型定义、会话管理、查询构造,每一层都要考虑同步和异步的差异。而Tortoise-ORM从设计之初就是异步原生的,它的模型定义风格类似Django ORM,查询接口直接返回可等待对象,和FastAPI的async/await模型天然契合。

1.1 同步ORM放在异步框架里有多别扭

说句实话,用FastAPI配SQLAlchemy的人不在少数,但这并不代表这两者配合得有多舒服。SQLAlchemy诞生于同步时代,它的核心架构围绕Connection、Session、UnitOfWork这些概念展开,在同步Web框架里确实强大。但到了FastAPI这种异步框架里,问题就来了:

  • 同步Session会阻塞事件循环,一旦数据库查询稍微慢一点,整个服务的并发能力立刻下降
  • 如果你用run_in_executor或者run_sync去规避阻塞,又引入了线程切换的开销和上下文管理的复杂度
  • Session的生命周期管理非常容易出错,请求结束忘记关闭连接,连接池瞬间被打满
  • 在依赖注入里获取Session、提交事务、回滚异常,每一步都要手写样板代码

我有一次在生产环境遇到过:一个简单的查询接口,数据库响应只要5毫秒,但加上SQLAlchemy的Session初始化、上下文切换、线程池调度之后,整个请求耗时飙到了30毫秒以上。后来换成Tortoise-ORM,同样查询降到8毫秒。这个差距在高并发场景下会被放大得非常明显。

1.2 Tortoise-ORM的核心特性:异步原生、Django式模型

Tortoise-ORM的设计思路非常直接:既然FastAPI整个生态都是async的,那ORM也应该是async的。它从底层就是用asyncpgaiosqlite这些异步驱动,查询操作返回awaitable对象,不需要任何同步转异步的桥接层。

模型定义则基本是Django ORM的复刻。用过Django的人上手Tortoise几乎零成本,模型字段、Meta选项、QuerySet链式调用、Manager机制都似曾相识。但和Django ORM不同的是,Tortoise不依赖全局设置,没有Django那种一大套配置体系的负担,可以像普通库一样嵌入任何异步框架。

我整理过一份对比表,方便你理解它的定位:

特性Tortoise-ORMSQLAlchemy 2.0
异步原生需额外用AsyncSession
模型定义风格Django式,简洁直观声明式,功能强大但冗余
学习成本中高
QuerySet链式查询支持支持
迁移工具AerichAlembic
适合场景快速开发REST API复杂业务逻辑、动态查询

当然SQLAlchemy在复杂查询、多数据库方言支持上仍然比Tortoise成熟,但如果你做一个纯REST接口的FastAPI项目,Tortoise-ORM几乎是为这个场景量身定做的。

2. 环境准备与项目结构:第一步就决定后面是否顺利

2.1 依赖安装

安装Tortoise-ORM基本就是两个包:

pip install tortoise-orm asyncpg

如果开发环境用SQLite,可以加一个aiosqlite。生产环境用PostgreSQL时,asyncpg是官方推荐的驱动,性能表现也明显优于psycopg2变体。

一个完整的requirements.txt大概是这样的:

fastapi uvicorn[standard] tortoise-orm asyncpg aerich pydantic pydantic-settings

其中aerich是Tortoise-ORM的数据库迁移工具,官方推荐配合使用,后面会专门讲。

2.2 目录结构参考

我在实际项目中比较常用的结构是这样的:

myproject/ ├── app/ │ ├── __init__.py │ ├── main.py # FastAPI应用入口 │ ├── config.py # 配置管理 │ ├── database.py # Tortoise初始化与连接配置 │ ├── models/ # ORM模型 │ │ ├── __init__.py │ │ ├── user.py │ │ ├── article.py │ │ └── category.py │ ├── schemas/ # Pydantic模型 │ │ ├── __init__.py │ │ ├── user.py │ │ └── article.py │ ├── routers/ # API路由 │ │ ├── __init__.py │ │ ├── user.py │ │ └── article.py │ └── services/ # 业务逻辑层 │ ├── __init__.py │ └── user_service.py ├── migrations/ # Aerich迁移文件 ├── .env └── requirements.txt

这个结构参考了Django和FastAPI社区的常见实践。modelsschemasroutersservices四层分离,各自职责清晰——模型管数据库结构,Pydantic模型管接口入参出参,路由管HTTP层,服务层放业务逻辑。

很多教程喜欢把所有代码堆在一个main.py里,演示可以,但真实项目千万别这么干。原因很简单:一旦模型多了、路由多了,单文件维护成本指数上涨,而且会导致循环导入这种莫名其妙的问题。我在后面的模型定义部分会再提这个坑。

2.3 配置初始化:register_tortoise还是手动init

Tortoise-ORM在FastAPI中集成有两种方式:

第一种是用官方提供的register_tortoise

from fastapi import FastAPI from tortoise.contrib.fastapi import register_tortoise app = FastAPI() register_tortoise( app, config={ "connections": { "default": "postgres://user:password@localhost:5432/mydb" }, "apps": { "models": { "models": ["app.models", "aerich.models"], "default_connection": "default", } }, }, generate_schemas=True, add_exception_handlers=True, )

register_tortoise会在FastAPI应用启动时自动调用Tortoise.init(),关闭时自动调用Tortoise.close_connections()。优点是省事,几乎零配置。

第二种是手动管理生命周期,用FastAPI的lifespan事件:

from contextlib import asynccontextmanager from fastapi import FastAPI @asynccontextmanager async def lifespan(app: FastAPI): await Tortoise.init( db_url="postgres://user:password@localhost:5432/mydb", modules={"models": ["app.models", "aerich.models"]} ) yield await Tortoise.close_connections() app = FastAPI(lifespan=lifespan)

我个人更推荐第二种。原因有两个:

第一,register_tortoise里的generate_schemas=True会在应用启动时自动建表,这在开发阶段很方便,但生产环境绝对不能用——一旦模型字段有变更,它会直接改表结构,非常危险。用Lifespan方案,你可以完全控制建表的时机,生产环境用Aerich迁移,开发环境才手动建表。

第二,Lifespan给了你一个注入额外初始化逻辑的位置。比如启动时初始化缓存连接、创建默认管理员账号、加载配置等。把数据库生命周期和这些逻辑放一起,整体更干净。

3. 模型定义:像写Django一样写ORM,但要留意几个不同点

Tortoise-ORM的模型定义确实非常Django化:

from tortoise import fields, models class User(models.Model): id = fields.IntField(pk=True) username = fields.CharField(max_length=64, unique=True, index=True) email = fields.CharField(max_length=255, unique=True) hashed_password = fields.CharField(max_length=128) is_active = fields.BooleanField(default=True) created_at = fields.DatetimeField(auto_now_add=True) updated_at = fields.DatetimeField(auto_now=True) class Meta: table = "users" ordering = ["-id"] def __str__(self): return self.username

可以看到字段类型、约束、默认值、Meta选项这些概念,和Django几乎一模一样。但如果你真按Django的习惯去写,会遇到几个Tortoise特有的事情:

3.1 字段类型与选项

Tortoise的字段类型覆盖了日常所需:IntFieldBigIntFieldCharFieldTextFieldBooleanFieldDatetimeFieldDateFieldFloatFieldDecimalFieldJSONFieldUUIDFieldForeignKeyFieldManyToManyFieldOneToOneField

JSON字段是Tortoise的亮点,直接对应PostgreSQL的jsonb类型:

class Article(models.Model): id = fields.IntField(pk=True) title = fields.CharField(max_length=200) metadata = fields.JSONField(default=dict)

这个字段在日常业务里太常用了——存一些不想单独建表的配置信息、扩展属性、埋点数据,都非常方便。

字段选项里要记住几个常用的:

  • pk=True指定主键
  • index=True加单列索引
  • unique=True唯一约束
  • null=True允许为空
  • default=xxx默认值
  • auto_now_add=True创建时自动写入当前时间
  • auto_now=True每次更新时自动写入当前时间

注意auto_now_addauto_now的区别:前者只在创建时写入一次,后者每次保存都会更新。两者都优先使用数据库时间还是Python时间?Tortoise默认在模型层面处理,也就是说它是取Python这边的当前时间。如果数据库服务器和应用服务器时间不同步,需要留意这一点。

3.2 关系字段的定义

关系字段是ORM的精华,Tortoise定义关系的方式如下:

class Category(models.Model): id = fields.IntField(pk=True) name = fields.CharField(max_length=64) class Article(models.Model): id = fields.IntField(pk=True) title = fields.CharField(max_length=200) content = fields.TextField() category = fields.ForeignKeyField( "models.Category", related_name="articles", on_delete=fields.CASCADE ) tags = fields.ManyToManyField( "models.Tag", related_name="articles", through="article_tags" )

这里有几个值得注意的细节:

第一个是ForeignKeyField的第一个参数是字符串"models.Category"。这是Tortoise的模块解析方式,models对应初始化时配置里的models模块名,Category是模型类名。很多新手在这里写错,导致报错找不到模型。注意这个字符串并不是Python的import路径,而是Tortoise内部注册的模块名。

第二个是related_name。它决定从Category反向获取文章时用的属性名。定义了related_name="articles"之后,你就可以通过category.articles查询该分类下的所有文章。如果没定义,Tortoise会生成一个默认名称。

第三个是on_delete选项。Tortoise支持fields.CASCADEfields.SET_NULLfields.RESTRICTfields.SET_DEFAULT等,行为和Django一致。注意:如果使用SET_NULL,对应字段必须设置null=True,否则保存时会报错。

3.3 模型导入规范:循环导入的根源

这是Tortoise项目里最常见的坑之一。当你的模型之间互相引用时,比如User和Article有外键关系,如果两个模型文件互相import,很容易触发循环导入错误。

我推荐的处理方式:在models/__init__.py里统一导入和暴露所有模型。

# app/models/__init__.py from app.models.user import User from app.models.article import Article from app.models.category import Category __all__ = ["User", "Article", "Category"]

这样模型文件内部不需要互相import。Tortoise初始化时指定"models": ["app.models", "aerich.models"],它会把app.models.__init__里暴露的所有模型加载进内部注册表。

如果实在需要在模型文件里引用另一个模型类(比如编写自定义方法),优先使用字符串引用——在方法内部再import,不要写在文件顶部。这个习惯能帮你避开99%的循环导入问题。

4. FastAPI集成细节:生命周期、依赖注入和连接池

4.1 连接管理和连接池的行为

Tortoise-ORM默认使用异步数据库驱动的连接池功能。以asyncpg为例,它会为每个数据库连接维护一个连接池,连接池的默认大小是10。这个值可以在db_url里配置:

postgres://user:password@localhost:5432/mydb?maxsize=20&minsize=5

注意这里的参数名是maxsizeminsize,放在数据库连接URL的query字符串里。maxsize定义连接池的上限。高并发场景下,如果连接池被打满,后续请求会等待空闲连接释放,这一点和任何连接池产品都类似。

我一般在配置里把maxsize设为CPU核心数×2左右,然后根据压测结果再调整。设置过大会造成数据库连接数溢出,设置过小则并发能力受限。另外,生产环境务必配置一个合理的连接空闲超时,避免数据库主动断开连接导致大量报错。

4.2 在依赖注入中获取连接和会话

Tortoise没有像SQLAlchemy那样强调Session的概念,它的操作基本都绑定在模型类上。所以依赖注入这部分不需要为数据库做太多事,最常见的模式是:

from fastapi import APIRouter, Depends, HTTPException from app.models import User from app.schemas.user import UserOut router = APIRouter(prefix="/users", tags=["users"]) async def get_user_or_404(user_id: int) -> User: user = await User.get_or_none(id=user_id) if not user: raise HTTPException(status_code=404, detail="User not found") return user @router.get("/{user_id}", response_model=UserOut) async def read_user(user: User = Depends(get_user_or_404)): return user

有一些教程对依赖注入谈虎色变,觉得每个接口都要写一遍依赖很啰嗦。其实FastAPI的依赖注入可以做减法:把通用逻辑(如权限校验、对象查询、分页参数)抽到依赖里,反而减少了重复代码。而且依赖可以叠加,比如先校验用户,再校验权限,再取资源,链路非常清晰。

4.3 请求级别的数据隔离

一个Tortoise使用中比较经典的问题是:多个请求并发修改同一个对象,会不会互相干扰?

答案是:Tortoise的模型实例没有全局共享Session,每次查询都从连接池获取一个连接,操作完成后释放。所以不会出现A请求改动了对象的属性,B请求随后也拿到了这个脏对象的情况。但要注意一点:从一个独立查询中获得的同一个数据库记录的两个Python实例,它们之间是没有同步的。比如:

user_a = await User.get(id=1) user_b = await User.get(id=1) await user_a.update_from_dict({"username": "new_name"}).save() # 此时user_b.username仍然是旧值

如果你在长时间运行的业务逻辑中持有模型实例,要留意这个特性。比如WebSocket长连接中,如果依赖模型实例的实时状态,就需要在每次使用前重新查询。

5. 核心CRUD写法和我在项目中踩过的那些坑

5.1 创建:灵活但需要小心的update_from_dict

创建用户最直观的方式是直接实例化:

user = User(username="admin", email="admin@example.com", hashed_password="...") await user.save()

批量创建用bulk_create

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

bulk_create在插入大量数据时性能优势明显,它会把多条INSERT合并,减少数据库往返。实验数据显示,插入1000条记录时,bulk_create比循环save快50倍以上。

另一个非常有用的方法是update_from_dict——它接收一个字典,只更新字典中的字段,适合和Pydantic的model_dump()配合使用:

user_data = payload.model_dump(exclude_unset=True) await user.update_from_dict(user_data).save()

exclude_unset=True很重要:它让Pydantic只在客户端显式传了字段时才把该字段放进model_dump()结果里。这样就不会把客户端没传的字段覆盖为空值。

5.2 查询:从get到复杂QuerySet

Tortoise查询接口非常顺手,日常CRUD基本就是这一套:

# 获取单条 user = await User.get(id=1) # 获取单条,不存在返回None user = await User.get_or_none(id=1) # 获取或创建 user, created = await User.get_or_create(username="admin") # 列表查询 users = await User.filter(is_active=True).order_by("-created_at").limit(10).offset(0) # 排除特定条件 users = await User.exclude(is_active=False) # 计数 count = await User.filter(is_active=True).count() # 存在性判断 exists = await User.filter(username="admin").exists() # 只取某些字段 names = await User.all().values_list("username", flat=True)

还有一个很有用的查询是in_bulk,按主键批量获取:

users = await User.in_bulk([1, 2, 3])

它返回一个以主键为键的字典,在需要按ID批量关联查询数据时非常高效。

5.3 更新与删除

更新常见的方式有以下几种:

# 方式一:先查后改 user = await User.get(id=1) user.username = "new_name" await user.save(update_fields=["username"]) # 方式二:update_from_dict await user.update_from_dict({"username": "new_name"}).save() # 方式三:QuerySet批量更新 await User.filter(id=1).update(username="new_name") # 方式四:批量更新 await User.filter(is_active=False).update(is_active=True)

update_fields参数可以控制只更新指定的字段,减少不必要的写操作。注意如果模型里有auto_now=True的字段,无论是否在update_fields中指定,它都会被更新。

删除操作:

# 单个删除 user = await User.get(id=1) await user.delete() # QuerySet批量删除 await User.filter(is_active=False).delete()

批量删除时有个容易踩的坑:如果你在外键关系上设置了on_delete=fields.RESTRICT,批量删除有被引用记录时,数据库会抛约束异常。所以在批量删除前,要么先处理关联数据,要么把外键设为CASCADE

5.4 分页和排序

FastAPI项目中,分页参数通常是pagepage_size,Tortoise查起来很简单:

page = 1 page_size = 20 users = await User.all().order_by("-id").offset((page - 1) * page_size).limit(page_size) total = await User.all().count()

排序字段注意以下几点:

  • 默认按主键升序排列,用-id表示降序
  • 多种排序条件用多个参数实现,如.order_by("category", "-created_at")
  • 关联字段排序:.order_by("category__name")可以对关联表的字段排序
  • 多字段、关联字段排序很灵活,但每次排序都会影响查询计划,字段多时记得加索引

6. 通过FastAPI依赖注入实现事务控制

6.1 为什么需要事务装饰器

Tortoise-ORM提供in_transaction上下文管理器来开启事务:

from tortoise.transactions import in_transaction async def transfer_money(from_user_id: int, to_user_id: int, amount: int): async with in_transaction() as conn: from_user = await User.get(id=from_user_id, using_db=conn) to_user = await User.get(id=to_user_id, using_db=conn) from_user.balance -= amount to_user.balance += amount await from_user.save(using_db=conn) await to_user.save(using_db=conn)

注意in_transaction上下文管理器返回的conn是一个连接对象,在事务块内的所有查询和保存都要显式传入using_db=conn。这是比较容易遗漏的地方——如果忘了传,查询就跑到事务外的连接上去了,整个事务的控制也就失效了。

我的建议是:模块内封装每个事务场景为一个独立函数,函数内部所有数据库操作都在同一事务里执行。然后在FastAPI路由层直接调用这个函数。

6.2 在依赖注入中实现事务的完整写法

再来看看依赖注入怎么配合。一个比较完整的例子是:创建订单时,同时更新商品库存和用户余额,这必须在同一个事务里完成。

from fastapi import Depends from tortoise.transactions import in_transaction from app.services.order_service import create_order_with_stock_change @router.post("/orders") async def create_order( order_data: OrderCreate, current_user: User = Depends(get_current_user), ): order = await create_order_with_stock_change( user_id=current_user.id, order_data=order_data ) return order

create_order_with_stock_change内部使用in_transaction

async def create_order_with_stock_change(user_id: int, order_data: OrderCreate): async with in_transaction() as conn: product = await Product.get(id=order_data.product_id, using_db=conn) if product.stock < order_data.quantity: raise BusinessError("库存不足") product.stock -= order_data.quantity await product.save(update_fields=["stock"], using_db=conn) order = await Order.create( user_id=user_id, product_id=product.id, quantity=order_data.quantity, total_price=product.price * order_data.quantity, using_db=conn ) return order

6.3 事务中常见的两个坑

第一个是异常处理。in_transaction默认在代码块异常退出时自动回滚。但如果你自己在代码块内捕获了异常并吞掉了,没有继续向上抛,事务会正常提交。这就意味着,业务上判断"库存不足"后,不能只记录日志然后继续往下走,必须通过raise把错误抛出去,否则后续的写操作会被执行。

第二个是长事务问题。一个事务内如果执行了太多次查询、等待了外部接口,数据库层面的连接会一直被占用。如果并发上来,连接池会被快速耗尽。我的实践是事务内只做必须原子化的写操作,把耗时较长的外部调用和复杂计算放在事务前后。这条建议在Django和SQLAlchemy项目中同样适用。

7. 关系查询与性能优化:prefetch_related和select_related

7.1 两种关系的查询方式

Tortoise-ORM用fetch_related来加载关联数据,Python风格还是Django那套,但名字不同。

先看一个经典场景:文章列表,每篇文章需要返回所属分类名称。

articles = await Article.all().prefetch_related("category")

prefetch_related会执行一次主查询,然后对关联对象做一次批量IN查询,把每条记录对应的关联对象缓存到模型实例上。它对ForeignKeyFieldManyToManyField都适用,是解决N+1问题的关键。

如果只需要关联表的一个字段,更节省性能的方式是select_related

articles = await Article.all().select_related("category")

select_related通过SQL JOIN把关联表数据一起查出来。对比prefetch_related,它减少了一次查询次数,但如果一张Article关联多个表,JOIN会显著增大结果集大小,增加网络传输和内存的开销。我通常是这样的取舍:

  • 需要主表数据完整、关联表字段少:用select_related
  • 关联表字段多、或是一对多/多对多关系:用prefetch_related

7.2 annotate聚合查询

聚合查询的需求很常见——比如统计每个分类下的文章数量。

from tortoise.functions import Count categories = await Category.annotate( article_count=Count("articles") ).order_by("-article_count")

这会给每个Category对象动态加上一个article_count属性。你可以在返回时直接使用。

Tortoise支持的聚合函数还有SumAvgMaxMin

from tortoise.functions import Sum total_sales = await Order.annotate( total=Sum("amount") ).group_by("product_id").values("product_id", "total")

7.3 我在一个真实接口里看到的N+1问题有多夸张

之前接手过一个老项目,文章列表接口返回50篇文章,写法的伪码大概是:

articles = await Article.all() for article in articles: category = await article.category # 每篇文章查一次数据库

这个逻辑本身很直观,但性能极其糟糕。50篇文章,每篇一次分类查询,再加上主查询,总共51次数据库往返。而且Tortoise的关联属性默认懒加载,访问article.category才会发起查询。用户请求一次列表,数据库忙活半天,日志刷得飞起。

prefetch_related改成两条SQL之后完全变了个样:

articles = await Article.all().prefetch_related("category")

之前那个接口在测试环境测出的平均延迟是180ms,改完后直接掉到15ms。数据库压力小了一个数量级。

如果你不确定自己的接口有没有N+1问题,可以在开发环境打开Tortoise的SQL日志,数一数一次请求发起了多少条SQL。也可以用Article.all().query()查看实际的SQL语句,这比凭感觉猜测靠谱得多。

8. 分页、序列化与Pydantic的配合

8.1 Tortoise-ORM和Pydantic的Model

Tortoise-ORM官方提供了一套Pydantic辅助函数,最常用的是pydantic_model_creator

from tortoise.contrib.pydantic import pydantic_model_creator from app.models.user import User UserOut = pydantic_model_creator(User, name="UserOut")

这样生成的Pydantic模型可以自动将Tortoise模型实例序列化为响应数据:

@router.get("/users/{user_id}", response_model=UserOut) async def get_user(user_id: int): user = await User.get(id=user_id) return user

但用过几次之后,我逐渐发现这种自动生成的模型有几个问题:

第一,它生成的字段类型、约束条件基于Tortoise字段,很可能不符合前端接口的需求。比如某些内部字段(如hashed_password)也会暴露出去,除非你用exclude排除:

UserOut = pydantic_model_creator( User, name="UserOut", exclude=["hashed_password"] )

第二,嵌套关系需要手动指定include,而且嵌套层的规则不好精确控制。

第三,定制输出格式时(比如日期格式、金额单位换算),在自动生成模型上调起来比较费劲。

所以我现在的实践是:开发阶段用pydantic_model_creator快速出接口,上线前如果接口结构复杂,就手工写Pydantic模型,配合一个手工的序列化方法或转换函数。手工模型的可控性强很多,代码也一眼能看懂。

8.2 分页返回结构的统一封装

做REST接口时,分页返回结构大家习惯各不相同,我自用一种"data + total + page + page_size"的格式:

from typing import Generic, TypeVar from pydantic import BaseModel T = TypeVar("T") class PaginatedResponse(BaseModel, Generic[T]): items: list[T] total: int page: int page_size: int async def paginated_query(request, model, filters=None, page=1, page_size=20): queryset = model.all() if filters: queryset = queryset.filter(**filters) total = await queryset.count() items = await queryset.offset((page - 1) * page_size).limit(page_size) return PaginatedResponse[model](...)

统一之后,前端处理分页数据只需解析一种结构。

8.3 序列化时的日期和时区问题

Tortoise默认使用UTC时间存储。如果API返回的时间字符串带有时区偏移,前端处理起来相对简单;但很多项目直接返回的是2025-01-01T10:00:00Z这种格式,前端解析时用的又是本地时区,就会出现"时间对不上"的bug。

我的习惯是接口层统一返回UTC的ISO8601格式,由前端负责转换为本地显示。这个约定需要在接口文档里写清楚,避免前后端各改各的。

9. 数据库迁移:不用Aerich,后面一定会后悔

9.1 Aerich是什么,为什么需要它

generate_schemas=True可以在开发时自动建表,但一旦上线,你就不能指望它了。模型字段一改动,比如新增一个字段、修改字段长度、加了一个索引,都需要同步到数据库,而且不能丢失已有数据。这就是数据库迁移工具存在的意义。

Aerich是Tortoise-ORM官方推荐的迁移工具,作用类似Alembic之于SQLAlchemy、makemigrations之于Django。它对比当前模型定义和数据库实际结构的差异,自动生成迁移SQL。

9.2 Aerich实战流程

Aerich需要在Tortoise初始化之后使用。常见流程分三步:

第一步:初始化

aerich init -t app.database.TORTOISE_CONFIG

TORTOISE_CONFIG是你在database.py里定义的配置字典:

TORTOISE_CONFIG = { "connections": { "default": DATABASE_URL }, "apps": { "models": { "models": ["app.models", "aerich.models"], "default_connection": "default", } } }

第二步:初始化数据库并生成初始迁移

aerich init-db

这会生成一个migrations目录,里面是0001_xxx.py迁移脚本,同时建立一张aerich表用于版本记录。

第三步:模型变更后生成并执行迁移

aerich migrate --name add_user_role aerich upgrade

migrate生成迁移脚本,upgrade执行。回滚用aerich downgrade

9.3 迁移过程容易踩的坑

迁移工具再智能也有解决不了的情况。我遇到过最典型的有三类:

第一类是字段类型变更。比如CharField改成TextField,大多数数据库可以平滑升级;但TextField改成CharField,如果数据长度超过新字段限制,数据库会直接报错。这类问题Aerich生成脚本时不会主动提醒,上线前一定要检查。

第二类是数据迁移。如果删除了一个字段或者修改了枚举值,迁移本身不会处理已有数据的转换,需要你写额外的数据迁移脚本,在upgrade前后手动执行。

第三类是迁移脚本的合并冲突。多人协作时,两个开发者各自加了字段,生成了两个迁移文件,upgrade时可能因为版本顺序问题报错。解决办法是尽量让migrate之前先upgrade到最新版本,再migrate;或者使用aerich--name参数尽量细化每次迁移的职责,减少冲突面。

10. 我项目里踩过的几个Tortoise-ORM实战坑

10.1 get_queryset返回的是QuerySet还是list

Tortoise的Model.all()返回的是QuerySet,注意它不是一个普通Python可迭代对象。很多新手会直接for item in await Model.all(),这没问题,但如果你把await Model.all()的结果直接当成list使用,比如取len()或用索引取值,就会出问题。

正确做法:

users = await User.all() # QuerySet,还没有执行查询 len(users) # 0,因为还没有取出数据 users[0] # 会报错 # 需要取出数据: users = list(await User.all())

10.2 外键字段的写入方式

给模型实例设置外键时,直接赋值对象或ID都可以:

article = Article(title="Hello") article.category = category_obj # 赋值模型实例 article.category_id = category_obj.id # 赋值ID await article.save()

categorycategory_id是Tortoise自动生成的两个属性,前者是模型实例,后者是ID。写操作时如果传入的是ID,直接用category_id=1即可,省去一次查询。

10.3 JSONField的序列化问题

Tortoise的JSONField底层是数据库原生JSON类型。如果你往里存一个普通dict,拿出来时会发现它是一个普通Python dict,不是Json包装类型,这算好消息。但如果你的数据里有datetime对象,序列化时要注意——数据库驱动可能无法直接处理datetime,会抛TypeError

我习惯在写入前做一次JSON安全的序列化,比如把所有datetime转成ISO字符串:

import datetime def json_safe(data): if isinstance(data, dict): return {k: json_safe(v) for k, v in data.items()} if isinstance(data, (list, tuple)): return [json_safe(item) for item in data] if isinstance(data, datetime.datetime): return data.isoformat() return data

10.4 时间字段和时区

再次强调时区问题。FastAPI项目通常服务全球用户,公共配置中设置时区时就需要注意。Tortoise-ORM默认使用UTC时间。如果在项目的settings里设置了:

TIMEZONE = "Asia/Shanghai"

Tortoise并不会自动把存储的所有时间都转换成本地时区。它存的还是UTC,只有在读取时由驱动转换。最省心的方案是:数据库统一UTC,接口层统一ISO8601带时区偏移,前端负责显示转换。

10.5 在unittest中使用内存数据库

Tortoise官方提供了tortoise.contrib.test模块,写单测时可用SQLite内存数据库:

import unittest from tortoise.contrib.test import IsolatedTestCase class TestUserModel(IsolatedTestCase): async def test_create_user(self): user = await User.create(username="test", email="test@example.com") self.assertEqual(user.username, "test")

IsolatedTestCase自动为每个测试方法提供独立的事务隔离环境,测完就回滚,不会污染数据库表。这个方案在CI里跑测试非常方便,而且速度比真实数据库快得多。

11. 最后再分享一点我实际使用的体会

Tortoise-ORM给我的整体感受是:上手快、写起来爽、坑也不算多,而且坑基本都有章可循。

我建议第一次用Tortoise-ORM的开发者别急着直接上项目,先花半天时间:

  • 用SQLite跑通一个最小的FastAPI应用
  • 定义两个有关联关系的模型
  • 写一遍完整CRUD
  • 配置Aerich做一次迁移

把这几步跑完,Tortoise-ORM的核心使用方式基本就掌握了。之后再接入PostgreSQL、写事务、做性能优化,这些都是在原有基础上的扩展。

我前后在三个FastAPI项目里用过Tortoise-ORM,从简单的博客系统到订单交易系统都试过。目前稳定跑在生产环境的最大规模是一个日请求量百万级的服务,数据库连接池和查询性能表现都符合预期。当然,如果项目复杂到需要动态拼接极其复杂的SQL、涉及特殊数据库类型、又依赖SQLAlchemy生态的插件,那Tortoise-ORM可能不是最优选。但在"FastAPI + 纯REST接口 + 快速迭代"这个典型场景里,它是我目前测试下来最顺手的选择。

需要专业的网站建设服务?

联系我们获取免费的网站建设咨询和方案报价,让我们帮助您实现业务目标

立即咨询