tortoise-orm,一个异步的 Python 库

大家好，我是木木。

今天给大家分享一个异步的 Python 库，tortoise-orm。

tortoise-orm

如果你的服务已经是 FastAPI、aiohttp 或其他 asyncio 技术栈，数据库层还在同步 ORM 和线程池之间来回折腾，tortoise-orm 会是一个更顺手的选择。它提供接近 Django ORM 的模型定义、过滤、排序、关系和事务能力，同时把执行链路放在 async/await 里，适合用来写中小型异步服务、后台任务和需要关系型数据库的 API。

项目地址：https://github.com/tortoise/tortoise-orm

官方文档：https://tortoise.github.io/

三大特点

异步原生

模型、查询和事务都可以直接 await，适合 asyncio Web 服务，避免同步数据库调用卡住事件循环。

写法熟悉

整体 API 接近 Django ORM，从字段、过滤到关系查询都比较直观，团队迁移成本低。

数据库多样

支持 SQLite、PostgreSQL、MySQL 等常见关系型数据库，不过生产环境仍要认真配置连接池和迁移工具。

最佳实践

安装方式很直接：pip install tortoise-orm。

第一个场景先看模型定义和基础查询。这段代码解决的问题是：用一个异步模型完成建表、写入、过滤和排序。

fromtortoiseimportTortoise,fields,models,run_asyncclassProduct(models.Model):id=fields.IntField(pk=True)name=fields.CharField(max_length=40)price=fields.IntField()stock=fields.IntField(default=0)asyncdefmain():awaitTortoise.init(db_url="sqlite://:memory:",modules={"models":["__main__"]})awaitTortoise.generate_schemas()awaitProduct.bulk_create([Product(name="Keyboard",price=80,stock=5),Product(name="Mouse",price=30,stock=10),Product(name="Monitor",price=220,stock=0),])items=awaitProduct.filter(stock__gt=0).order_by("-price")print("available:",len(items))foriteminitems:print(item.name,item.price,item.stock)awaitTortoise.close_connections()run_async(main())

第二个场景是关系和聚合。业务里经常需要按作者、项目、订单这类关系做统计，Tortoise 可以把外键关系和聚合函数放在同一条查询里。

fromtortoiseimportTortoise,fields,models,run_asyncfromtortoise.functionsimportCount,SumclassAuthor(models.Model):id=fields.IntField(pk=True)name=fields.CharField(max_length=40)classBook(models.Model):id=fields.IntField(pk=True)title=fields.CharField(max_length=80)pages=fields.IntField()author=fields.ForeignKeyField("models.Author",related_name="books")asyncdefmain():awaitTortoise.init(db_url="sqlite://:memory:",modules={"models":["__main__"]})awaitTortoise.generate_schemas()alice=awaitAuthor.create(name="Alice")bob=awaitAuthor.create(name="Bob")awaitBook.bulk_create([Book(author=alice,title="Async ORM",pages=180),Book(author=alice,title="Query Guide",pages=220),Book(author=bob,title="SQLite Tips",pages=90),])rows=awaitAuthor.annotate(total=Count("books"),pages=Sum("books__pages")).order_by("-pages")forrowinrows:print(row.name,row.total,row.pages)awaitTortoise.close_connections()run_async(main())

进阶一点看事务。异步服务里更容易出现多个请求同时写入的问题，涉及余额、库存、唯一约束时，要把关键写操作放进事务边界。

fromtortoiseimportTortoise,fields,models,run_asyncfromtortoise.exceptionsimportIntegrityErrorfromtortoise.transactionsimportin_transactionclassAccount(models.Model):id=fields.IntField(pk=True)name=fields.CharField(max_length=40,unique=True)balance=fields.IntField(default=0)asyncdefmain():awaitTortoise.init(db_url="sqlite://:memory:",modules={"models":["__main__"]})awaitTortoise.generate_schemas()awaitAccount.create(name="alice",balance=100)try:asyncwithin_transaction():awaitAccount.create(name="bob",balance=50)awaitAccount.create(name="alice",balance=200)exceptIntegrityErrorasexc:print("rollback:",exc.__class__.__name__)accounts=awaitAccount.all().order_by("name")print("accounts:",len(accounts))print("names:",", ".join(a.nameforainaccounts))awaitTortoise.close_connections()run_async(main())

环境与版本信息

本文示例使用 Python 3.11，tortoise-orm 版本为 1.1.7，关键依赖包括 aiosqlite 0.21.0 和 pypika-tortoise 0.6.5。当前包声明 Python 版本要求为 3.10 及以上。

适用场景

适合已经采用 asyncio 的 API 服务、需要关系型数据库但不想写大量 SQL 的中小项目，以及希望用接近 Django ORM 风格管理模型和关系的团队。

不适用场景

如果项目主要是同步 Django，或者已经深度依赖 SQLAlchemy 的高级表达式、复杂迁移和生态插件，就不一定需要切换。超复杂查询场景也要评估原生 SQL 的可维护性。

上线检查

1. 配置真实数据库连接池，不要沿用内存 SQLite。
2. 使用 Aerich 或团队已有方案管理迁移。
3. 给慢查询、唯一约束和事务失败补充测试。

总结

tortoise-orm 的价值在于：用熟悉的 ORM 写法服务异步应用。它不是最重的 ORM，但在 async API 项目里足够轻、顺、好落地。