TortoiseORM

锦秀

TortoiseORM全面指南：从基础到高级应用目录简介安装与配置基本用法高级特性性能优化测试策略与其他库的集成最佳实践常见问题及解决方案实际应用场景与其他ORM的比较结语简介TortoiseORM是一个强大的异步PythonORM，专为现代异步Web框架设计。它结合了DjangoORM的直观性和异步编程的高效性，为开发者提供了一个理想的数据库交互解决方案。主要特性完全异步操作类似Django的API强类型支持复杂关系处理丰富的查询API事务支持多数据库支持（SQLite,PostgreSQL,MySQL）安装与配置基本安装pipinstalltortoise-orm1数据库特定安装pipinstalltortoise-orm[asyncpg]#PostgreSQLpipinstalltortoise-orm[aiomysql]#MySQL12基本配置fromtortoiseimportTortoise,run_asyncasyncdefinit():awaitTortoise.init(db_url='sqlite://db.sqlite3',modules={'models':['myapp.models']})awaitTortoise.generate_schemas()run_async(init())12345678910基本用法模型定义fromtortoiseimportModel,fieldsclassUser(Model):id=fields.IntField(pk=True)username=fields.CharField(max_length=50,unique=True)email=fields.CharField(max_length=100,unique=True)password=fields.CharField(max_length=128)is_active=fields.BooleanField(default=True)created_at=fields.DatetimeField(auto_now_add=True)def__str__(self):returnself.usernameclassPost(Model):id=fields.IntField(pk=True)title=fields.CharField(max_length=200)content=fields.TextField()author=fields.ForeignKeyField('models.User',related_name='posts')created_at=fields.DatetimeField(auto_now_add=True)updated_at=fields.DatetimeField(auto_now=True)def__str__(self):returnself.title1234567891011121314151617181920212223CRUD操作#创建user=awaitUser.create(username='johndoe',email='john@example.com',password='securepass')#读取user=awaitUser.get(username='johndoe')all_users=awaitUser.all()#更新awaitUser.filter(id=1).update(is_active=False)#删除awaitUser.filter(id=1).delete()123456789101112高级特性复杂查询fromtortoise.expressionsimportQ#组合查询active_users=awaitUser.filter(Q(is_active=True)&(Q(username__contains='john')|Q(email__contains='john')))#排除non_admin_users=awaitUser.exclude(username='admin')#排序sorted_users=awaitUser.all().order_by('-created_at','username')#限制结果recent_users=awaitUser.all().order_by('-created_at').limit(5)123456789101112131415关系查询#预加载关联对象users_with_posts=awaitUser.all().prefetch_related('posts')#反向关系查询user=awaitUser.get(id=1)user_posts=awaituser.posts.all()#多对多关系classTag(Model):name=fields.CharField(max_length=50)posts=fields.ManyToManyField('models.Post',related_name='tags')#添加多对多关系post=awaitPost.get(id=1)tag=awaitTag.get(id=1)awaitpost.tags.add(tag)#查询多对多关系posts_with_tag=awaitPost.filter(tags__id=1)12345678910111213141516171819聚合和注解fromtortoise.functionsimportCount,Avg,Sum#计算每个用户的帖子数users_post_count=awaitUser.annotate(post_count=Count('posts'))#获取帖子数量最多的用户most_active_user=awaitUser.annotate(post_count=Count('posts')).order_by('-post_count').first()#计算平均帖子长度avg_post_length=awaitPost.annotate(avg_length=Avg('content__length')).values('avg_length')12345678910原生SQL#执行原生SQL查询users=awaitUser.raw("SELECT*FROMuserWHEREis_active=1")#执行原生SQL命令awaitTortoise.get_connection('default').execute_query("UPDATEuserSETis_active=1WHEREid=1")12345性能优化批量操作#批量创建users_data=[{"username":f"user{i}","email":f"user{i}@example.com"}foriinrange(100)]awaitUser.bulk_create([User(**data)fordatainusers_data])#批量更新awaitUser.filter(is_active=False).update(is_active=True)123456789索引优化classOptimizedPost(Model):title=fields.CharField(max_length=200,index=True)content=fields.TextField()created_at=fields.DatetimeField(auto_now_add=True,index=True)classMeta:indexes=[("title","created_at")]1234567连接池配置awaitTortoise.init(db_url='postgres://user:pass@localhost:5432/dbname',modules={'models':['myapp.models']},connections={'default':{'engine':'tortoise.backends.asyncpg','credentials':{'host':'localhost','port':'5432','user':'user','password':'pass','database':'dbname','minsize':1,'maxsize':10,}}})123456789101112131415161718测试策略使用测试数据库fromtortoise.contrib.testimportinitializer,finalizerdefsetUpModule():initializer(['myapp.models'],db_url='sqlite://:memory:')deftearDownModule():finalizer()classTestUser(TestCase):asyncdeftest_create_user(self):user=awaitUser.create(username='testuser',email='test@example.com')self.assertEqual(awaitUser.all().count(),1)self.assertEqual(user.username,'testuser')12345678910111213事务测试fromtortoise.transactionsimportin_transactionasyncdeftest_transaction():asyncwithin_transaction():user=awaitUser.create(username='transactionuser',email='trans@example.com')#事务会在这里回滚，不会实际创建用户assertawaitUser.filter(username='transactionuser').exists()==False1234567与其他库的集成与FastAPI集成fromfastapiimportFastAPIfromtortoise.contrib.fastapiimportregister_tortoiseapp=FastAPI()register_tortoise(app,db_url="sqlite://db.sqlite3",modules={"models":["myapp.models"]},generate_schemas=True,add_exception_handlers=True,)@app.get("/users")asyncdefget_users():returnawaitUser.all()12345678910111213141516与Pydantic集成fromtortoise.contrib.pydanticimportpydantic_model_creatorUserPydantic=pydantic_model_creator(User)@app.post("/users")asyncdefcreate_user(user:UserPydantic):user_obj=awaitUser.create(**user.dict(exclude_unset=True))returnawaitUserPydantic.from_tortoise_orm(user_obj)12345678最佳实践使用异步上下文管理器：在处理长时间运行的操作时，使用异步上下文管理器来确保资源被正确释放。合理使用预加载：使用prefetch_related来减少数据库查询次数，但要避免过度预加载。批量操作：对于大量数据的操作，使用批量创建、更新或删除方法。使用事务：对于需要保证一致性的复杂操作，使用事务来确保原子性。索引优化：根据查询模式合理设置索引，但要避免过度索引。异步迭代：处理大量数据时，使用异步迭代器来减少内存使用。asyncforuserinUser.all():#处理每个用户12定期清理连接池：在长时间运行的应用中，定期重置数据库连接池以防止连接泄漏。常见问题及解决方案循环导入问题：解决方案：使用字符串模型引用或延迟导入。大量数据的内存问题：解决方案：使用limit和offset进行分页查询，或使用异步迭代器。复杂查询性能问题：解决方案：使用select_related和prefetch_related优化查询，必要时使用原生SQL。多数据库支持：解决方案：使用Tortoise的多数据库配置功能，为不同模型指定不同的数据库。实际应用场景博客系统：使用User、Post、Comment和Tag模型，展示复杂的关系查询和内容管理。电子商务平台：实现Product、Order、Customer模型，展示事务处理和库存管理。社交媒体应用：使用User、Post、Like、Follow模型，展示高并发场景下的性能优化。内容管理系统(CMS)：实现灵活的内容类型和字段定义，展示动态模型创建和管理。与其他ORM的比较vsSQLAlchemy：TortoiseORM专注于异步，API设计更简洁。SQLAlchemy功能更全面，生态系统更大。vsDjangoORM：TortoiseORM支持异步操作，更适合现代异步框架。DjangoORM与Django框架深度集成，提供更多开箱即用的功能。vsPeewee：TortoiseORM原生支持异步操作。Peewee更轻量，但主要用于同步操作。结语TortoiseORM为Python异步编程提供了一个强大而灵活的ORM解决方案。它结合了直观的API设计和高效的异步操作，使得在现代Web应用中处理数据库操作变得简单而高效。无论是构建小型项目还是大型应用，TortoiseORM都能提供必要的工具和灵活性。随着异步编程在Python社区中的日益普及，TortoiseORM的重要性可能会进一步提升。持续关注其发展和最佳实践，将有助于构建更高效、更可扩展的Python应用。