跳至主要內容

Tortoise-ORM异步数据库操作

Li JiaXin教程后端框架测试开发Tortoise-ORM异步数据库大约 10 分钟

Tortoise-ORM异步数据库操作

如果说Peewee是数据库操作的"自行车"（轻便好用），那么Tortoise-ORM就是"特斯拉"（现代化、高性能、智能化）。Tortoise-ORM专为异步Python设计，让测试开发工程师能够在高并发场景下进行高效的数据库操作，就像给测试平台装上了"涡轮增压器"！

一、Tortoise-ORM简介：异步数据库操作的艺术

什么是Tortoise-ORM？

Tortoise-ORM是一个受Django ORM启发的异步ORM框架，专为Python的asyncio设计：

核心特性：

  • 异步优先：原生支持async/await语法
  • 高性能：基于asyncio，支持高并发
  • Django风格：熟悉的API设计，学习成本低
  • 类型安全：支持Python类型提示
  • 多数据库：支持PostgreSQL、MySQL、SQLite

Tortoise-ORM vs Peewee

特性Tortoise-ORMPeewee测试平台适用性
异步支持⭐⭐⭐⭐⭐❌高并发测试场景
性能⭐⭐⭐⭐⭐⭐⭐⭐大量数据处理
学习难度⭐⭐⭐⭐⭐需要异步编程基础
生态系统⭐⭐⭐⭐⭐⭐⭐快速发展中
FastAPI集成⭐⭐⭐⭐⭐⭐⭐⭐完美匹配

二、环境搭建与配置

安装Tortoise-ORM

# 基础安装
pip install tortoise-orm

# 包含数据库驱动
pip install tortoise-orm[asyncpg]  # PostgreSQL
pip install tortoise-orm[aiomysql] # MySQL
pip install tortoise-orm[aiosqlite] # SQLite

# 完整安装（推荐）
pip install tortoise-orm[asyncpg,aiomysql,aiosqlite]

基础配置

# config.py
TORTOISE_ORM = {
    "connections": {
        # SQLite配置（开发环境）
        "default": {
            "engine": "tortoise.backends.sqlite",
            "credentials": {
                "file_path": "test_platform.db"
            }
        },
        
        # PostgreSQL配置（生产环境）
        "postgres": {
            "engine": "tortoise.backends.asyncpg",
            "credentials": {
                "host": "localhost",
                "port": "5432",
                "user": "postgres",
                "password": "password",
                "database": "test_platform",
                "minsize": 1,
                "maxsize": 10,
            }
        },
        
        # MySQL配置
        "mysql": {
            "engine": "tortoise.backends.aiomysql",
            "credentials": {
                "host": "localhost",
                "port": 3306,
                "user": "root",
                "password": "password",
                "database": "test_platform",
                "minsize": 1,
                "maxsize": 10,
            }
        }
    },
    "apps": {
        "models": {
            "models": ["models", "aerich.models"],
            "default_connection": "default",
        }
    }
}

初始化Tortoise-ORM

# main.py
from tortoise import Tortoise
import asyncio

async def init_db():
    """初始化数据库连接"""
    await Tortoise.init(
        db_url='sqlite://test_platform.db',
        modules={'models': ['models']}
    )
    # 生成数据库表
    await Tortoise.generate_schemas()

async def close_db():
    """关闭数据库连接"""
    await Tortoise.close_connections()

# 在应用启动时调用
if __name__ == "__main__":
    asyncio.run(init_db())

三、模型定义：构建数据结构

基础模型定义

# models.py
from tortoise.models import Model
from tortoise import fields
from datetime import datetime
from enum import Enum

class Priority(str, Enum):
    """优先级枚举"""
    LOW = "low"
    MEDIUM = "medium"
    HIGH = "high"
    CRITICAL = "critical"

class TestStatus(str, Enum):
    """测试状态枚举"""
    PENDING = "pending"
    RUNNING = "running"
    PASSED = "passed"
    FAILED = "failed"
    SKIPPED = "skipped"

class BaseModel(Model):
    """基础模型类"""
    id = fields.IntField(pk=True)
    created_at = fields.DatetimeField(auto_now_add=True)
    updated_at = fields.DatetimeField(auto_now=True)
    
    class Meta:
        abstract = True

class Project(BaseModel):
    """项目模型"""
    name = fields.CharField(max_length=100, unique=True, description="项目名称")
    description = fields.TextField(null=True, description="项目描述")
    status = fields.CharField(max_length=20, default="active", description="项目状态")
    
    # 反向关系
    test_suites: fields.ReverseRelation["TestSuite"]
    
    def __str__(self):
        return self.name
    
    class Meta:
        table = "projects"
        table_description = "测试项目表"

class TestSuite(BaseModel):
    """测试套件模型"""
    name = fields.CharField(max_length=100, description="套件名称")
    description = fields.TextField(null=True, description="套件描述")
    
    # 外键关系
    project = fields.ForeignKeyField(
        "models.Project", 
        related_name="test_suites",
        description="所属项目"
    )
    
    # 反向关系
    test_cases: fields.ReverseRelation["TestCase"]
    
    def __str__(self):
        return f"{self.project.name} - {self.name}"
    
    class Meta:
        table = "test_suites"
        unique_together = (("name", "project"),)  # 项目内套件名唯一

class TestCase(BaseModel):
    """测试用例模型"""
    name = fields.CharField(max_length=200, description="用例名称")
    description = fields.TextField(null=True, description="用例描述")
    
    # 外键关系
    test_suite = fields.ForeignKeyField(
        "models.TestSuite",
        related_name="test_cases",
        description="所属测试套件"
    )
    
    # 枚举字段
    priority = fields.CharEnumField(Priority, default=Priority.MEDIUM, description="优先级")
    status = fields.CharEnumField(TestStatus, default=TestStatus.PENDING, description="执行状态")
    
    # 测试内容
    url = fields.CharField(max_length=500, description="测试URL")
    method = fields.CharField(max_length=10, default="GET", description="HTTP方法")
    headers = fields.JSONField(default=dict, description="请求头")
    body = fields.JSONField(null=True, description="请求体")
    expected_status = fields.IntField(default=200, description="期望状态码")
    
    # 执行结果
    actual_status = fields.IntField(null=True, description="实际状态码")
    response_time = fields.FloatField(null=True, description="响应时间(ms)")
    response_body = fields.TextField(null=True, description="响应内容")
    error_message = fields.TextField(null=True, description="错误信息")
    
    # 统计信息
    run_count = fields.IntField(default=0, description="执行次数")
    success_count = fields.IntField(default=0, description="成功次数")
    
    # 反向关系
    execution_logs: fields.ReverseRelation["ExecutionLog"]
    
    @property
    def success_rate(self) -> float:
        """成功率"""
        if self.run_count == 0:
            return 0.0
        return self.success_count / self.run_count
    
    def __str__(self):
        return f"{self.test_suite.name} - {self.name}"
    
    class Meta:
        table = "test_cases"
        ordering = ["-created_at"]

class ExecutionLog(BaseModel):
    """执行日志模型"""
    test_case = fields.ForeignKeyField(
        "models.TestCase",
        related_name="execution_logs",
        description="关联测试用例"
    )
    
    status = fields.CharEnumField(TestStatus, description="执行状态")
    start_time = fields.DatetimeField(description="开始时间")
    end_time = fields.DatetimeField(null=True, description="结束时间")
    duration = fields.FloatField(null=True, description="执行时长(ms)")
    
    request_data = fields.JSONField(null=True, description="请求数据")
    response_data = fields.JSONField(null=True, description="响应数据")
    error_message = fields.TextField(null=True, description="错误信息")
    
    # 环境信息
    executor = fields.CharField(max_length=100, null=True, description="执行者")
    environment = fields.CharField(max_length=50, default="test", description="执行环境")
    
    class Meta:
        table = "execution_logs"
        ordering = ["-start_time"]

# 多对多关系示例
class Tag(BaseModel):
    """标签模型"""
    name = fields.CharField(max_length=50, unique=True, description="标签名称")
    color = fields.CharField(max_length=7, default="#1890ff", description="标签颜色")
    
    # 多对多关系
    test_cases = fields.ManyToManyField(
        "models.TestCase",
        related_name="tags",
        through="test_case_tags"  # 自定义中间表名
    )
    
    class Meta:
        table = "tags"

字段类型详解

# 常用字段类型示例
class FieldExamples(BaseModel):
    # 基础字段
    char_field = fields.CharField(max_length=100)
    text_field = fields.TextField()
    int_field = fields.IntField()
    float_field = fields.FloatField()
    bool_field = fields.BooleanField(default=False)
    
    # 时间字段
    date_field = fields.DateField()
    datetime_field = fields.DatetimeField()
    time_field = fields.TimeField()
    
    # 特殊字段
    json_field = fields.JSONField()
    uuid_field = fields.UUIDField()
    decimal_field = fields.DecimalField(max_digits=10, decimal_places=2)
    
    # 约束字段
    unique_field = fields.CharField(max_length=50, unique=True)
    indexed_field = fields.CharField(max_length=50, index=True)
    nullable_field = fields.CharField(max_length=50, null=True)
    default_field = fields.CharField(max_length=50, default="default_value")
    
    # 关系字段
    foreign_key = fields.ForeignKeyField("models.Project")
    one_to_one = fields.OneToOneField("models.User")
    many_to_many = fields.ManyToManyField("models.Tag")

四、基础CRUD操作

创建记录

# 创建单个记录
async def create_project():
    """创建项目"""
    project = await Project.create(
        name="API测试项目",
        description="用于API接口测试的项目",
        status="active"
    )
    print(f"创建项目: {project.name}, ID: {project.id}")
    return project

# 批量创建
async def create_multiple_projects():
    """批量创建项目"""
    projects_data = [
        {"name": "项目1", "description": "描述1"},
        {"name": "项目2", "description": "描述2"},
        {"name": "项目3", "description": "描述3"},
    ]
    
    projects = await Project.bulk_create([
        Project(**data) for data in projects_data
    ])
    print(f"批量创建了 {len(projects)} 个项目")
    return projects

# 获取或创建
async def get_or_create_project(name: str):
    """获取或创建项目"""
    project, created = await Project.get_or_create(
        name=name,
        defaults={"description": f"{name}的描述"}
    )
    
    if created:
        print(f"创建了新项目: {project.name}")
    else:
        print(f"项目已存在: {project.name}")
    
    return project

### 查询记录

```python
# 基础查询
async def basic_queries():
    """基础查询示例"""

    # 获取所有记录
    all_projects = await Project.all()
    print(f"所有项目数量: {len(all_projects)}")

    # 获取单个记录
    project = await Project.get(name="API测试项目")
    print(f"项目: {project.name}")

    # 获取单个记录（可能不存在）
    project = await Project.get_or_none(name="不存在的项目")
    if project:
        print(f"找到项目: {project.name}")
    else:
        print("项目不存在")

    # 获取第一个记录
    first_project = await Project.first()
    if first_project:
        print(f"第一个项目: {first_project.name}")

# 条件查询
async def conditional_queries():
    """条件查询示例"""

    # 简单条件
    active_projects = await Project.filter(status="active")

    # 多个条件（AND）
    recent_projects = await Project.filter(
        status="active",
        created_at__gte=datetime.now() - timedelta(days=30)
    )

    # 排除条件
    non_active_projects = await Project.exclude(status="active")

    # 复杂条件
    from tortoise.expressions import Q

    complex_projects = await Project.filter(
        Q(status="active") | Q(status="pending"),
        name__icontains="测试"
    )

# 查询操作符
async def query_operators():
    """查询操作符示例"""

    # 字符串操作
    projects = await Project.filter(
        name__icontains="API",      # 包含
        name__istartswith="测试",    # 开头
        name__iendswith="项目",      # 结尾
        name__iexact="API测试项目"   # 精确匹配（忽略大小写）
    )

    # 数值操作
    test_cases = await TestCase.filter(
        run_count__gt=10,           # 大于
        run_count__gte=5,           # 大于等于
        run_count__lt=100,          # 小于
        run_count__lte=50,          # 小于等于
        run_count__in=[10, 20, 30], # 在列表中
        run_count__range=(5, 50)    # 范围
    )

    # 时间操作
    recent_cases = await TestCase.filter(
        created_at__year=2024,
        created_at__month=3,
        created_at__day=15,
        updated_at__date=datetime.now().date()
    )

    # 空值检查
    cases_with_error = await TestCase.filter(error_message__isnull=False)
    cases_without_error = await TestCase.filter(error_message__isnull=True)

# 排序和限制
async def ordering_and_limiting():
    """排序和限制示例"""

    # 排序
    projects = await Project.all().order_by("name")  # 升序
    projects = await Project.all().order_by("-created_at")  # 降序
    projects = await Project.all().order_by("status", "-created_at")  # 多字段

    # 限制数量
    latest_projects = await Project.all().order_by("-created_at").limit(10)

    # 分页
    page_size = 20
    page = 2
    offset = (page - 1) * page_size

    projects = await Project.all().offset(offset).limit(page_size)

    # 计数
    total_count = await Project.all().count()
    active_count = await Project.filter(status="active").count()

# 聚合查询
async def aggregation_queries():
    """聚合查询示例"""
    from tortoise.functions import Count, Sum, Avg, Max, Min

    # 统计每个项目的测试套件数量
    projects_with_suite_count = await Project.annotate(
        suite_count=Count("test_suites")
    ).all()

    for project in projects_with_suite_count:
        print(f"{project.name}: {project.suite_count} 个测试套件")

    # 统计测试用例的执行情况
    execution_stats = await TestCase.all().aggregate(
        total_runs=Sum("run_count"),
        avg_success_rate=Avg("success_count"),
        max_runs=Max("run_count"),
        min_runs=Min("run_count")
    )

    print(f"总执行次数: {execution_stats['total_runs']}")
    print(f"平均成功次数: {execution_stats['avg_success_rate']}")

### 更新记录

```python
# 更新单个记录
async def update_single_record():
    """更新单个记录"""
    project = await Project.get(name="API测试项目")
    project.description = "更新后的描述"
    project.status = "updated"
    await project.save()

    # 或者使用update_fields指定更新字段
    await project.save(update_fields=["description", "status"])

# 批量更新
async def bulk_update():
    """批量更新"""
    # 更新所有符合条件的记录
    updated_count = await Project.filter(status="pending").update(status="active")
    print(f"更新了 {updated_count} 个项目状态")

    # 使用F表达式进行字段运算
    from tortoise.expressions import F

    # 将所有测试用例的运行次数加1
    await TestCase.all().update(run_count=F("run_count") + 1)

# 原子更新
async def atomic_update():
    """原子更新操作"""
    from tortoise.transactions import atomic

    @atomic()
    async def update_test_result(test_case_id: int, success: bool):
        test_case = await TestCase.get(id=test_case_id)
        test_case.run_count += 1
        if success:
            test_case.success_count += 1
            test_case.status = TestStatus.PASSED
        else:
            test_case.status = TestStatus.FAILED

        await test_case.save()

        # 记录执行日志
        await ExecutionLog.create(
            test_case=test_case,
            status=test_case.status,
            start_time=datetime.now(),
            end_time=datetime.now(),
            duration=100.5
        )

### 删除记录

```python
# 删除单个记录
async def delete_single_record():
    """删除单个记录"""
    project = await Project.get(name="要删除的项目")
    await project.delete()

# 批量删除
async def bulk_delete():
    """批量删除"""
    # 删除所有符合条件的记录
    deleted_count = await TestCase.filter(status=TestStatus.FAILED).delete()
    print(f"删除了 {deleted_count} 个失败的测试用例")

# 级联删除
async def cascade_delete():
    """级联删除示例"""
    # 删除项目时，相关的测试套件和测试用例也会被删除
    project = await Project.get(name="要删除的项目")

    # 先获取相关数据统计
    suite_count = await project.test_suites.all().count()
    case_count = await TestCase.filter(test_suite__project=project).count()

    print(f"即将删除项目: {project.name}")
    print(f"包含 {suite_count} 个测试套件, {case_count} 个测试用例")

    await project.delete()
    print("删除完成")

## 五、关系查询与事务管理

### 预加载关系数据

```python
# 避免N+1查询问题
async def efficient_relationship_queries():
    """高效的关系查询"""

    # 使用prefetch_related预加载一对多关系
    projects = await Project.all().prefetch_related("test_suites", "test_suites__test_cases")

    for project in projects:
        print(f"项目: {project.name}")
        for suite in project.test_suites:
            print(f"  套件: {suite.name}")
            for case in suite.test_cases:
                print(f"    用例: {case.name}")

    # 使用select_related预加载外键关系
    test_cases = await TestCase.all().select_related("test_suite", "test_suite__project")

    for case in test_cases:
        # 这些访问不会产生额外的数据库查询
        print(f"用例: {case.name}")
        print(f"套件: {case.test_suite.name}")
        print(f"项目: {case.test_suite.project.name}")

### 事务管理

```python
from tortoise.transactions import atomic, in_transaction

# 装饰器方式
@atomic()
async def create_project_with_suite(project_name: str, suite_name: str):
    """创建项目和测试套件（事务）"""
    project = await Project.create(name=project_name)
    suite = await TestSuite.create(name=suite_name, project=project)
    return project, suite

# 上下文管理器方式
async def transfer_test_cases(from_suite_id: int, to_suite_id: int, case_ids: list):
    """转移测试用例（事务）"""
    async with in_transaction() as conn:
        from_suite = await TestSuite.get(id=from_suite_id).using_db(conn)
        to_suite = await TestSuite.get(id=to_suite_id).using_db(conn)

        # 更新测试用例的所属套件
        await TestCase.filter(id__in=case_ids).using_db(conn).update(
            test_suite=to_suite
        )

        print(f"成功转移 {len(case_ids)} 个测试用例")

六、FastAPI集成实战

完整的FastAPI应用示例

# main.py
from fastapi import FastAPI, HTTPException, Depends
from tortoise.contrib.fastapi import register_tortoise
from pydantic import BaseModel
from typing import List, Optional

app = FastAPI(title="测试平台API")

# Pydantic模型
class ProjectCreate(BaseModel):
    name: str
    description: Optional[str] = None

class ProjectResponse(BaseModel):
    id: int
    name: str
    description: Optional[str]
    status: str
    created_at: datetime

    class Config:
        orm_mode = True

# API端点
@app.post("/projects", response_model=ProjectResponse)
async def create_project(project: ProjectCreate):
    """创建项目"""
    try:
        db_project = await Project.create(**project.dict())
        return ProjectResponse.from_orm(db_project)
    except Exception as e:
        raise HTTPException(status_code=400, detail=str(e))

@app.get("/projects", response_model=List[ProjectResponse])
async def get_projects(skip: int = 0, limit: int = 100):
    """获取项目列表"""
    projects = await Project.all().offset(skip).limit(limit)
    return [ProjectResponse.from_orm(project) for project in projects]

@app.get("/projects/{project_id}", response_model=ProjectResponse)
async def get_project(project_id: int):
    """获取单个项目"""
    project = await Project.get_or_none(id=project_id)
    if not project:
        raise HTTPException(status_code=404, detail="项目不存在")
    return ProjectResponse.from_orm(project)

# 注册Tortoise-ORM
register_tortoise(
    app,
    db_url="sqlite://test_platform.db",
    modules={"models": ["models"]},
    generate_schemas=True,
    add_exception_handlers=True,
)

七、总结与最佳实践

通过这篇文章，我们深入学习了Tortoise-ORM的核心特性：

✅ 异步数据库操作：高性能的异步CRUD操作 ✅ 模型设计：灵活的字段类型和关系定义 ✅ 查询优化：预加载和聚合查询技巧 ✅ 事务管理：确保数据一致性的事务操作 ✅ FastAPI集成：完美的现代化Web开发体验

最佳实践总结

  1. 模型设计原则

    • 使用枚举类型提高数据一致性
    • 合理设计索引提升查询性能
    • 使用抽象基类减少重复代码

  2. 查询优化技巧

    • 使用select_related预加载外键关系
    • 使用prefetch_related预加载一对多关系
    • 避免在循环中进行数据库查询

  3. 事务使用建议

    • 保持事务尽可能短小
    • 在事务中避免长时间的外部调用
    • 合理使用嵌套事务

  4. 性能优化要点

    • 使用批量操作减少数据库往返
    • 合理使用数据库索引
    • 监控慢查询并优化

🎯 下一步预告：掌握了Tortoise-ORM的异步数据库操作，下一篇我们将学习数据库迁移工具Aerich，了解如何管理数据库版本和结构变更！

Tortoise-ORM让我们体验到了异步数据库操作的强大威力，让我们继续深入学习数据库管理的高级技巧！🚀