尧图网站开发 
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的。它从底层就是用asyncpg、aiosqlite这些异步驱动查询操作返回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社区的常见实践。models、schemas、routers、services四层分离各自职责清晰——模型管数据库结构Pydantic模型管接口入参出参路由管HTTP层服务层放业务逻辑。很多教程喜欢把所有代码堆在一个main.py里演示可以但真实项目千万别这么干。原因很简单一旦模型多了、路由多了单文件维护成本指数上涨而且会导致循环导入这种莫名其妙的问题。我在后面的模型定义部分会再提这个坑。2.3 配置初始化register_tortoise还是手动initTortoise-ORM在FastAPI中集成有两种方式第一种是用官方提供的register_tortoisefrom fastapi import FastAPI from tortoise.contrib.fastapi import register_tortoise app FastAPI() register_tortoise( app, config{ connections: { default: postgres://user:passwordlocalhost:5432/mydb }, apps: { models: { models: [app.models, aerich.models], default_connection: default, } }, }, generate_schemasTrue, add_exception_handlersTrue, )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_urlpostgres://user:passwordlocalhost:5432/mydb, modules{models: [app.models, aerich.models]} ) yield await Tortoise.close_connections() app FastAPI(lifespanlifespan)我个人更推荐第二种。原因有两个第一register_tortoise里的generate_schemasTrue会在应用启动时自动建表这在开发阶段很方便但生产环境绝对不能用——一旦模型字段有变更它会直接改表结构非常危险。用Lifespan方案你可以完全控制建表的时机生产环境用Aerich迁移开发环境才手动建表。第二Lifespan给了你一个注入额外初始化逻辑的位置。比如启动时初始化缓存连接、创建默认管理员账号、加载配置等。把数据库生命周期和这些逻辑放一起整体更干净。3. 模型定义像写Django一样写ORM但要留意几个不同点Tortoise-ORM的模型定义确实非常Django化from tortoise import fields, models class User(models.Model): id fields.IntField(pkTrue) username fields.CharField(max_length64, uniqueTrue, indexTrue) email fields.CharField(max_length255, uniqueTrue) hashed_password fields.CharField(max_length128) is_active fields.BooleanField(defaultTrue) created_at fields.DatetimeField(auto_now_addTrue) updated_at fields.DatetimeField(auto_nowTrue) class Meta: table users ordering [-id] def __str__(self): return self.username可以看到字段类型、约束、默认值、Meta选项这些概念和Django几乎一模一样。但如果你真按Django的习惯去写会遇到几个Tortoise特有的事情3.1 字段类型与选项Tortoise的字段类型覆盖了日常所需IntField、BigIntField、CharField、TextField、BooleanField、DatetimeField、DateField、FloatField、DecimalField、JSONField、UUIDField、ForeignKeyField、ManyToManyField、OneToOneField。JSON字段是Tortoise的亮点直接对应PostgreSQL的jsonb类型class Article(models.Model): id fields.IntField(pkTrue) title fields.CharField(max_length200) metadata fields.JSONField(defaultdict)这个字段在日常业务里太常用了——存一些不想单独建表的配置信息、扩展属性、埋点数据都非常方便。字段选项里要记住几个常用的pkTrue指定主键indexTrue加单列索引uniqueTrue唯一约束nullTrue允许为空defaultxxx默认值auto_now_addTrue创建时自动写入当前时间auto_nowTrue每次更新时自动写入当前时间注意auto_now_add和auto_now的区别前者只在创建时写入一次后者每次保存都会更新。两者都优先使用数据库时间还是Python时间Tortoise默认在模型层面处理也就是说它是取Python这边的当前时间。如果数据库服务器和应用服务器时间不同步需要留意这一点。3.2 关系字段的定义关系字段是ORM的精华Tortoise定义关系的方式如下class Category(models.Model): id fields.IntField(pkTrue) name fields.CharField(max_length64) class Article(models.Model): id fields.IntField(pkTrue) title fields.CharField(max_length200) content fields.TextField() category fields.ForeignKeyField( models.Category, related_namearticles, on_deletefields.CASCADE ) tags fields.ManyToManyField( models.Tag, related_namearticles, througharticle_tags )这里有几个值得注意的细节第一个是ForeignKeyField的第一个参数是字符串models.Category。这是Tortoise的模块解析方式models对应初始化时配置里的models模块名Category是模型类名。很多新手在这里写错导致报错找不到模型。注意这个字符串并不是Python的import路径而是Tortoise内部注册的模块名。第二个是related_name。它决定从Category反向获取文章时用的属性名。定义了related_namearticles之后你就可以通过category.articles查询该分类下的所有文章。如果没定义Tortoise会生成一个默认名称。第三个是on_delete选项。Tortoise支持fields.CASCADE、fields.SET_NULL、fields.RESTRICT、fields.SET_DEFAULT等行为和Django一致。注意如果使用SET_NULL对应字段必须设置nullTrue否则保存时会报错。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:passwordlocalhost:5432/mydb?maxsize20minsize5注意这里的参数名是maxsize和minsize放在数据库连接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(iduser_id) if not user: raise HTTPException(status_code404, detailUser not found) return user router.get(/{user_id}, response_modelUserOut) 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(id1) user_b await User.get(id1) await user_a.update_from_dict({username: new_name}).save() # 此时user_b.username仍然是旧值如果你在长时间运行的业务逻辑中持有模型实例要留意这个特性。比如WebSocket长连接中如果依赖模型实例的实时状态就需要在每次使用前重新查询。5. 核心CRUD写法和我在项目中踩过的那些坑5.1 创建灵活但需要小心的update_from_dict创建用户最直观的方式是直接实例化user User(usernameadmin, emailadminexample.com, hashed_password...) await user.save()批量创建用bulk_createusers [User(usernamefuser{i}, emailfuser{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_unsetTrue) await user.update_from_dict(user_data).save()exclude_unsetTrue很重要它让Pydantic只在客户端显式传了字段时才把该字段放进model_dump()结果里。这样就不会把客户端没传的字段覆盖为空值。5.2 查询从get到复杂QuerySetTortoise查询接口非常顺手日常CRUD基本就是这一套# 获取单条 user await User.get(id1) # 获取单条不存在返回None user await User.get_or_none(id1) # 获取或创建 user, created await User.get_or_create(usernameadmin) # 列表查询 users await User.filter(is_activeTrue).order_by(-created_at).limit(10).offset(0) # 排除特定条件 users await User.exclude(is_activeFalse) # 计数 count await User.filter(is_activeTrue).count() # 存在性判断 exists await User.filter(usernameadmin).exists() # 只取某些字段 names await User.all().values_list(username, flatTrue)还有一个很有用的查询是in_bulk按主键批量获取users await User.in_bulk([1, 2, 3])它返回一个以主键为键的字典在需要按ID批量关联查询数据时非常高效。5.3 更新与删除更新常见的方式有以下几种# 方式一先查后改 user await User.get(id1) 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(id1).update(usernamenew_name) # 方式四批量更新 await User.filter(is_activeFalse).update(is_activeTrue)update_fields参数可以控制只更新指定的字段减少不必要的写操作。注意如果模型里有auto_nowTrue的字段无论是否在update_fields中指定它都会被更新。删除操作# 单个删除 user await User.get(id1) await user.delete() # QuerySet批量删除 await User.filter(is_activeFalse).delete()批量删除时有个容易踩的坑如果你在外键关系上设置了on_deletefields.RESTRICT批量删除有被引用记录时数据库会抛约束异常。所以在批量删除前要么先处理关联数据要么把外键设为CASCADE。5.4 分页和排序FastAPI项目中分页参数通常是page和page_sizeTortoise查起来很简单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(idfrom_user_id, using_dbconn) to_user await User.get(idto_user_id, using_dbconn) from_user.balance - amount to_user.balance amount await from_user.save(using_dbconn) await to_user.save(using_dbconn)注意in_transaction上下文管理器返回的conn是一个连接对象在事务块内的所有查询和保存都要显式传入using_dbconn。这是比较容易遗漏的地方——如果忘了传查询就跑到事务外的连接上去了整个事务的控制也就失效了。我的建议是模块内封装每个事务场景为一个独立函数函数内部所有数据库操作都在同一事务里执行。然后在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_idcurrent_user.id, order_dataorder_data ) return order在create_order_with_stock_change内部使用in_transactionasync def create_order_with_stock_change(user_id: int, order_data: OrderCreate): async with in_transaction() as conn: product await Product.get(idorder_data.product_id, using_dbconn) if product.stock order_data.quantity: raise BusinessError(库存不足) product.stock - order_data.quantity await product.save(update_fields[stock], using_dbconn) order await Order.create( user_iduser_id, product_idproduct.id, quantityorder_data.quantity, total_priceproduct.price * order_data.quantity, using_dbconn ) return order6.3 事务中常见的两个坑第一个是异常处理。in_transaction默认在代码块异常退出时自动回滚。但如果你自己在代码块内捕获了异常并吞掉了没有继续向上抛事务会正常提交。这就意味着业务上判断库存不足后不能只记录日志然后继续往下走必须通过raise把错误抛出去否则后续的写操作会被执行。第二个是长事务问题。一个事务内如果执行了太多次查询、等待了外部接口数据库层面的连接会一直被占用。如果并发上来连接池会被快速耗尽。我的实践是事务内只做必须原子化的写操作把耗时较长的外部调用和复杂计算放在事务前后。这条建议在Django和SQLAlchemy项目中同样适用。7. 关系查询与性能优化prefetch_related和select_related7.1 两种关系的查询方式Tortoise-ORM用fetch_related来加载关联数据Python风格还是Django那套但名字不同。先看一个经典场景文章列表每篇文章需要返回所属分类名称。articles await Article.all().prefetch_related(category)prefetch_related会执行一次主查询然后对关联对象做一次批量IN查询把每条记录对应的关联对象缓存到模型实例上。它对ForeignKeyField和ManyToManyField都适用是解决N1问题的关键。如果只需要关联表的一个字段更节省性能的方式是select_relatedarticles await Article.all().select_related(category)select_related通过SQL JOIN把关联表数据一起查出来。对比prefetch_related它减少了一次查询次数但如果一张Article关联多个表JOIN会显著增大结果集大小增加网络传输和内存的开销。我通常是这样的取舍需要主表数据完整、关联表字段少用select_related关联表字段多、或是一对多/多对多关系用prefetch_related7.2 annotate聚合查询聚合查询的需求很常见——比如统计每个分类下的文章数量。from tortoise.functions import Count categories await Category.annotate( article_countCount(articles) ).order_by(-article_count)这会给每个Category对象动态加上一个article_count属性。你可以在返回时直接使用。Tortoise支持的聚合函数还有Sum、Avg、Max、Minfrom tortoise.functions import Sum total_sales await Order.annotate( totalSum(amount) ).group_by(product_id).values(product_id, total)7.3 我在一个真实接口里看到的N1问题有多夸张之前接手过一个老项目文章列表接口返回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。数据库压力小了一个数量级。如果你不确定自己的接口有没有N1问题可以在开发环境打开Tortoise的SQL日志数一数一次请求发起了多少条SQL。也可以用Article.all().query()查看实际的SQL语句这比凭感觉猜测靠谱得多。8. 分页、序列化与Pydantic的配合8.1 Tortoise-ORM和Pydantic的ModelTortoise-ORM官方提供了一套Pydantic辅助函数最常用的是pydantic_model_creatorfrom tortoise.contrib.pydantic import pydantic_model_creator from app.models.user import User UserOut pydantic_model_creator(User, nameUserOut)这样生成的Pydantic模型可以自动将Tortoise模型实例序列化为响应数据router.get(/users/{user_id}, response_modelUserOut) async def get_user(user_id: int): user await User.get(iduser_id) return user但用过几次之后我逐渐发现这种自动生成的模型有几个问题第一它生成的字段类型、约束条件基于Tortoise字段很可能不符合前端接口的需求。比如某些内部字段如hashed_password也会暴露出去除非你用exclude排除UserOut pydantic_model_creator( User, nameUserOut, 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, filtersNone, page1, page_size20): 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_schemasTrue可以在开发时自动建表但一旦上线你就不能指望它了。模型字段一改动比如新增一个字段、修改字段长度、加了一个索引都需要同步到数据库而且不能丢失已有数据。这就是数据库迁移工具存在的意义。Aerich是Tortoise-ORM官方推荐的迁移工具作用类似Alembic之于SQLAlchemy、makemigrations之于Django。它对比当前模型定义和数据库实际结构的差异自动生成迁移SQL。9.2 Aerich实战流程Aerich需要在Tortoise初始化之后使用。常见流程分三步第一步初始化aerich init -t app.database.TORTOISE_CONFIGTORTOISE_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 upgrademigrate生成迁移脚本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还是listTortoise的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(titleHello) article.category category_obj # 赋值模型实例 article.category_id category_obj.id # 赋值ID await article.save()category和category_id是Tortoise自动生成的两个属性前者是模型实例后者是ID。写操作时如果传入的是ID直接用category_id1即可省去一次查询。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 data10.4 时间字段和时区再次强调时区问题。FastAPI项目通常服务全球用户公共配置中设置时区时就需要注意。Tortoise-ORM默认使用UTC时间。如果在项目的settings里设置了TIMEZONE Asia/ShanghaiTortoise并不会自动把存储的所有时间都转换成本地时区。它存的还是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(usernametest, emailtestexample.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接口 快速迭代这个典型场景里它是我目前测试下来最顺手的选择。 
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的。它从底层就是用asyncpg、aiosqlite这些异步驱动查询操作返回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社区的常见实践。models、schemas、routers、services四层分离各自职责清晰——模型管数据库结构Pydantic模型管接口入参出参路由管HTTP层服务层放业务逻辑。很多教程喜欢把所有代码堆在一个main.py里演示可以但真实项目千万别这么干。原因很简单一旦模型多了、路由多了单文件维护成本指数上涨而且会导致循环导入这种莫名其妙的问题。我在后面的模型定义部分会再提这个坑。2.3 配置初始化register_tortoise还是手动initTortoise-ORM在FastAPI中集成有两种方式第一种是用官方提供的register_tortoisefrom fastapi import FastAPI from tortoise.contrib.fastapi import register_tortoise app FastAPI() register_tortoise( app, config{ connections: { default: postgres://user:passwordlocalhost:5432/mydb }, apps: { models: { models: [app.models, aerich.models], default_connection: default, } }, }, generate_schemasTrue, add_exception_handlersTrue, )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_urlpostgres://user:passwordlocalhost:5432/mydb, modules{models: [app.models, aerich.models]} ) yield await Tortoise.close_connections() app FastAPI(lifespanlifespan)我个人更推荐第二种。原因有两个第一register_tortoise里的generate_schemasTrue会在应用启动时自动建表这在开发阶段很方便但生产环境绝对不能用——一旦模型字段有变更它会直接改表结构非常危险。用Lifespan方案你可以完全控制建表的时机生产环境用Aerich迁移开发环境才手动建表。第二Lifespan给了你一个注入额外初始化逻辑的位置。比如启动时初始化缓存连接、创建默认管理员账号、加载配置等。把数据库生命周期和这些逻辑放一起整体更干净。3. 模型定义像写Django一样写ORM但要留意几个不同点Tortoise-ORM的模型定义确实非常Django化from tortoise import fields, models class User(models.Model): id fields.IntField(pkTrue) username fields.CharField(max_length64, uniqueTrue, indexTrue) email fields.CharField(max_length255, uniqueTrue) hashed_password fields.CharField(max_length128) is_active fields.BooleanField(defaultTrue) created_at fields.DatetimeField(auto_now_addTrue) updated_at fields.DatetimeField(auto_nowTrue) class Meta: table users ordering [-id] def __str__(self): return self.username可以看到字段类型、约束、默认值、Meta选项这些概念和Django几乎一模一样。但如果你真按Django的习惯去写会遇到几个Tortoise特有的事情3.1 字段类型与选项Tortoise的字段类型覆盖了日常所需IntField、BigIntField、CharField、TextField、BooleanField、DatetimeField、DateField、FloatField、DecimalField、JSONField、UUIDField、ForeignKeyField、ManyToManyField、OneToOneField。JSON字段是Tortoise的亮点直接对应PostgreSQL的jsonb类型class Article(models.Model): id fields.IntField(pkTrue) title fields.CharField(max_length200) metadata fields.JSONField(defaultdict)这个字段在日常业务里太常用了——存一些不想单独建表的配置信息、扩展属性、埋点数据都非常方便。字段选项里要记住几个常用的pkTrue指定主键indexTrue加单列索引uniqueTrue唯一约束nullTrue允许为空defaultxxx默认值auto_now_addTrue创建时自动写入当前时间auto_nowTrue每次更新时自动写入当前时间注意auto_now_add和auto_now的区别前者只在创建时写入一次后者每次保存都会更新。两者都优先使用数据库时间还是Python时间Tortoise默认在模型层面处理也就是说它是取Python这边的当前时间。如果数据库服务器和应用服务器时间不同步需要留意这一点。3.2 关系字段的定义关系字段是ORM的精华Tortoise定义关系的方式如下class Category(models.Model): id fields.IntField(pkTrue) name fields.CharField(max_length64) class Article(models.Model): id fields.IntField(pkTrue) title fields.CharField(max_length200) content fields.TextField() category fields.ForeignKeyField( models.Category, related_namearticles, on_deletefields.CASCADE ) tags fields.ManyToManyField( models.Tag, related_namearticles, througharticle_tags )这里有几个值得注意的细节第一个是ForeignKeyField的第一个参数是字符串models.Category。这是Tortoise的模块解析方式models对应初始化时配置里的models模块名Category是模型类名。很多新手在这里写错导致报错找不到模型。注意这个字符串并不是Python的import路径而是Tortoise内部注册的模块名。第二个是related_name。它决定从Category反向获取文章时用的属性名。定义了related_namearticles之后你就可以通过category.articles查询该分类下的所有文章。如果没定义Tortoise会生成一个默认名称。第三个是on_delete选项。Tortoise支持fields.CASCADE、fields.SET_NULL、fields.RESTRICT、fields.SET_DEFAULT等行为和Django一致。注意如果使用SET_NULL对应字段必须设置nullTrue否则保存时会报错。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:passwordlocalhost:5432/mydb?maxsize20minsize5注意这里的参数名是maxsize和minsize放在数据库连接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(iduser_id) if not user: raise HTTPException(status_code404, detailUser not found) return user router.get(/{user_id}, response_modelUserOut) 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(id1) user_b await User.get(id1) await user_a.update_from_dict({username: new_name}).save() # 此时user_b.username仍然是旧值如果你在长时间运行的业务逻辑中持有模型实例要留意这个特性。比如WebSocket长连接中如果依赖模型实例的实时状态就需要在每次使用前重新查询。5. 核心CRUD写法和我在项目中踩过的那些坑5.1 创建灵活但需要小心的update_from_dict创建用户最直观的方式是直接实例化user User(usernameadmin, emailadminexample.com, hashed_password...) await user.save()批量创建用bulk_createusers [User(usernamefuser{i}, emailfuser{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_unsetTrue) await user.update_from_dict(user_data).save()exclude_unsetTrue很重要它让Pydantic只在客户端显式传了字段时才把该字段放进model_dump()结果里。这样就不会把客户端没传的字段覆盖为空值。5.2 查询从get到复杂QuerySetTortoise查询接口非常顺手日常CRUD基本就是这一套# 获取单条 user await User.get(id1) # 获取单条不存在返回None user await User.get_or_none(id1) # 获取或创建 user, created await User.get_or_create(usernameadmin) # 列表查询 users await User.filter(is_activeTrue).order_by(-created_at).limit(10).offset(0) # 排除特定条件 users await User.exclude(is_activeFalse) # 计数 count await User.filter(is_activeTrue).count() # 存在性判断 exists await User.filter(usernameadmin).exists() # 只取某些字段 names await User.all().values_list(username, flatTrue)还有一个很有用的查询是in_bulk按主键批量获取users await User.in_bulk([1, 2, 3])它返回一个以主键为键的字典在需要按ID批量关联查询数据时非常高效。5.3 更新与删除更新常见的方式有以下几种# 方式一先查后改 user await User.get(id1) 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(id1).update(usernamenew_name) # 方式四批量更新 await User.filter(is_activeFalse).update(is_activeTrue)update_fields参数可以控制只更新指定的字段减少不必要的写操作。注意如果模型里有auto_nowTrue的字段无论是否在update_fields中指定它都会被更新。删除操作# 单个删除 user await User.get(id1) await user.delete() # QuerySet批量删除 await User.filter(is_activeFalse).delete()批量删除时有个容易踩的坑如果你在外键关系上设置了on_deletefields.RESTRICT批量删除有被引用记录时数据库会抛约束异常。所以在批量删除前要么先处理关联数据要么把外键设为CASCADE。5.4 分页和排序FastAPI项目中分页参数通常是page和page_sizeTortoise查起来很简单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(idfrom_user_id, using_dbconn) to_user await User.get(idto_user_id, using_dbconn) from_user.balance - amount to_user.balance amount await from_user.save(using_dbconn) await to_user.save(using_dbconn)注意in_transaction上下文管理器返回的conn是一个连接对象在事务块内的所有查询和保存都要显式传入using_dbconn。这是比较容易遗漏的地方——如果忘了传查询就跑到事务外的连接上去了整个事务的控制也就失效了。我的建议是模块内封装每个事务场景为一个独立函数函数内部所有数据库操作都在同一事务里执行。然后在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_idcurrent_user.id, order_dataorder_data ) return order在create_order_with_stock_change内部使用in_transactionasync def create_order_with_stock_change(user_id: int, order_data: OrderCreate): async with in_transaction() as conn: product await Product.get(idorder_data.product_id, using_dbconn) if product.stock order_data.quantity: raise BusinessError(库存不足) product.stock - order_data.quantity await product.save(update_fields[stock], using_dbconn) order await Order.create( user_iduser_id, product_idproduct.id, quantityorder_data.quantity, total_priceproduct.price * order_data.quantity, using_dbconn ) return order6.3 事务中常见的两个坑第一个是异常处理。in_transaction默认在代码块异常退出时自动回滚。但如果你自己在代码块内捕获了异常并吞掉了没有继续向上抛事务会正常提交。这就意味着业务上判断库存不足后不能只记录日志然后继续往下走必须通过raise把错误抛出去否则后续的写操作会被执行。第二个是长事务问题。一个事务内如果执行了太多次查询、等待了外部接口数据库层面的连接会一直被占用。如果并发上来连接池会被快速耗尽。我的实践是事务内只做必须原子化的写操作把耗时较长的外部调用和复杂计算放在事务前后。这条建议在Django和SQLAlchemy项目中同样适用。7. 关系查询与性能优化prefetch_related和select_related7.1 两种关系的查询方式Tortoise-ORM用fetch_related来加载关联数据Python风格还是Django那套但名字不同。先看一个经典场景文章列表每篇文章需要返回所属分类名称。articles await Article.all().prefetch_related(category)prefetch_related会执行一次主查询然后对关联对象做一次批量IN查询把每条记录对应的关联对象缓存到模型实例上。它对ForeignKeyField和ManyToManyField都适用是解决N1问题的关键。如果只需要关联表的一个字段更节省性能的方式是select_relatedarticles await Article.all().select_related(category)select_related通过SQL JOIN把关联表数据一起查出来。对比prefetch_related它减少了一次查询次数但如果一张Article关联多个表JOIN会显著增大结果集大小增加网络传输和内存的开销。我通常是这样的取舍需要主表数据完整、关联表字段少用select_related关联表字段多、或是一对多/多对多关系用prefetch_related7.2 annotate聚合查询聚合查询的需求很常见——比如统计每个分类下的文章数量。from tortoise.functions import Count categories await Category.annotate( article_countCount(articles) ).order_by(-article_count)这会给每个Category对象动态加上一个article_count属性。你可以在返回时直接使用。Tortoise支持的聚合函数还有Sum、Avg、Max、Minfrom tortoise.functions import Sum total_sales await Order.annotate( totalSum(amount) ).group_by(product_id).values(product_id, total)7.3 我在一个真实接口里看到的N1问题有多夸张之前接手过一个老项目文章列表接口返回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。数据库压力小了一个数量级。如果你不确定自己的接口有没有N1问题可以在开发环境打开Tortoise的SQL日志数一数一次请求发起了多少条SQL。也可以用Article.all().query()查看实际的SQL语句这比凭感觉猜测靠谱得多。8. 分页、序列化与Pydantic的配合8.1 Tortoise-ORM和Pydantic的ModelTortoise-ORM官方提供了一套Pydantic辅助函数最常用的是pydantic_model_creatorfrom tortoise.contrib.pydantic import pydantic_model_creator from app.models.user import User UserOut pydantic_model_creator(User, nameUserOut)这样生成的Pydantic模型可以自动将Tortoise模型实例序列化为响应数据router.get(/users/{user_id}, response_modelUserOut) async def get_user(user_id: int): user await User.get(iduser_id) return user但用过几次之后我逐渐发现这种自动生成的模型有几个问题第一它生成的字段类型、约束条件基于Tortoise字段很可能不符合前端接口的需求。比如某些内部字段如hashed_password也会暴露出去除非你用exclude排除UserOut pydantic_model_creator( User, nameUserOut, 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, filtersNone, page1, page_size20): 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_schemasTrue可以在开发时自动建表但一旦上线你就不能指望它了。模型字段一改动比如新增一个字段、修改字段长度、加了一个索引都需要同步到数据库而且不能丢失已有数据。这就是数据库迁移工具存在的意义。Aerich是Tortoise-ORM官方推荐的迁移工具作用类似Alembic之于SQLAlchemy、makemigrations之于Django。它对比当前模型定义和数据库实际结构的差异自动生成迁移SQL。9.2 Aerich实战流程Aerich需要在Tortoise初始化之后使用。常见流程分三步第一步初始化aerich init -t app.database.TORTOISE_CONFIGTORTOISE_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 upgrademigrate生成迁移脚本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还是listTortoise的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(titleHello) article.category category_obj # 赋值模型实例 article.category_id category_obj.id # 赋值ID await article.save()category和category_id是Tortoise自动生成的两个属性前者是模型实例后者是ID。写操作时如果传入的是ID直接用category_id1即可省去一次查询。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 data10.4 时间字段和时区再次强调时区问题。FastAPI项目通常服务全球用户公共配置中设置时区时就需要注意。Tortoise-ORM默认使用UTC时间。如果在项目的settings里设置了TIMEZONE Asia/ShanghaiTortoise并不会自动把存储的所有时间都转换成本地时区。它存的还是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(usernametest, emailtestexample.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接口 快速迭代这个典型场景里它是我目前测试下来最顺手的选择。