Python实用工具：tortoise-orm入门到实战教程

作者：

在

tortoise-orm是一款专为异步Python应用设计的ORM（对象关系映射）工具，灵感源自Django ORM，支持异步数据库操作，兼容多种数据库（MySQL、PostgreSQL、SQLite等）。其工作原理是将Python类映射为数据库表，通过异步API执行CRUD操作，避免阻塞事件循环。优点是语法简洁、异步性能优、支持迁移；缺点是生态较SQLAlchemy小，部分复杂查询需手写SQL。License为Apache License 2.0。

一、tortoise-orm安装与环境配置

1.1 安装tortoise-orm

tortoise-orm支持pip直接安装，同时需根据使用的数据库安装对应的异步驱动。以常用的MySQL和SQLite为例：

安装核心库

pip install tortoise-orm

安装数据库驱动
SQLite（无需额外驱动，Python内置）
MySQL：安装asyncmy驱动

  pip install asyncmy

PostgreSQL：安装asyncpg驱动

  pip install asyncpg

1.2 验证安装

安装完成后，可通过以下代码验证是否安装成功：

import tortoise
print(f"tortoise-orm版本：{tortoise.__version__}")

运行代码，若输出版本号则说明安装成功。

二、tortoise-orm核心概念与初始化

2.1 核心概念

tortoise-orm的核心概念与Django ORM类似，主要包括：

Model：Python类，对应数据库中的一张表，类属性对应表字段。
Field：字段类型，如IntField、CharField、DatetimeField等，定义表字段的属性。
Manager：模型的查询管理器，通过objects属性提供查询方法（如all()、filter()）。
异步会话：所有数据库操作均为异步，需通过asyncio运行。

2.2 数据库初始化

使用tortoise-orm前，需先初始化数据库连接，通过configure方法配置连接信息，再调用init_models加载模型。

import asyncio
from tortoise import Tortoise, run_async
from tortoise.models import Model
from tortoise import fields

# 定义示例模型（后续详细讲解）
class User(Model):
    id = fields.IntField(pk=True)
    name = fields.CharField(max_length=50)
    age = fields.IntField(default=0)
    created_at = fields.DatetimeField(auto_now_add=True)

# 初始化函数
async def init_db():
    # 配置数据库连接
    await Tortoise.init(
        db_url="sqlite://test.db",  # SQLite数据库文件
        modules={"models": ["__main__"]}  # 模型所在模块
    )
    # 生成数据库表（首次运行时执行）
    await Tortoise.generate_schemas()

# 运行异步初始化
if __name__ == "__main__":
    run_async(init_db())

代码说明：

db_url：数据库连接字符串，格式为数据库类型://用户名:密码@地址:端口/数据库名，SQLite直接指定文件路径。
modules：指定包含模型的模块，__main__表示当前模块。
generate_schemas()：自动创建模型对应的数据库表，生产环境建议使用迁移工具。

三、tortoise-orm模型定义与字段类型

3.1 模型定义规则

tortoise-orm的模型需继承自tortoise.models.Model，每个模型类对应一张数据库表，表名默认是模型类名的小写复数形式（可通过Meta类自定义）。

from tortoise import fields
from tortoise.models import Model

class User(Model):
    # 主键字段，pk=True表示为主键
    id = fields.IntField(pk=True)
    # 字符串字段，max_length为必填参数
    username = fields.CharField(max_length=30, unique=True, description="用户名")
    # 密码字段，可设置默认值
    password = fields.CharField(max_length=100, default="123456")
    # 整数字段，设置默认值
    age = fields.IntField(default=0, description="年龄")
    # 布尔字段
    is_active = fields.BooleanField(default=True, description="是否激活")
    # 时间字段，auto_now_add=True表示创建时自动填充当前时间
    created_at = fields.DatetimeField(auto_now_add=True, description="创建时间")
    # 时间字段，auto_now=True表示更新时自动填充当前时间
    updated_at = fields.DatetimeField(auto_now=True, description="更新时间")

    class Meta:
        # 自定义表名
        table = "user"
        # 索引，可提升查询效率
        indexes = [("username",)]

代码说明：

pk=True：标记字段为主键，若未定义主键，tortoise-orm会自动创建一个名为id的自增主键。
unique=True：设置字段值唯一，避免重复数据。
description：字段描述，可选参数。
Meta类：用于配置模型的元数据，如自定义表名、索引、外键约束等。

3.2 常用字段类型

tortoise-orm提供了丰富的字段类型，满足不同数据存储需求，常用字段如下表所示：

| 字段类型 | 作用 | 常用参数 |
|-||-|
| IntField | 存储整数 | default、null |
| CharField | 存储字符串 | max_length、unique、default |
| TextField | 存储长文本 | null、default |
| DatetimeField | 存储日期时间 | auto_now_add、auto_now |
| BooleanField | 存储布尔值 | default |
| FloatField | 存储浮点数 | default、null |
| ForeignKeyField | 外键关联 | model_name、on_delete |

四、tortoise-orm核心操作：CRUD实战

CRUD是数据库操作的核心（创建、读取、更新、删除），tortoise-orm的所有操作均为异步，需在async函数中执行。

4.1 数据创建（Create）

向数据库中添加数据有两种方式：create()方法和save()方法。

方法1：使用`create()`直接创建

async def create_user():
    # 初始化数据库
    await init_db()
    # 创建单个用户
    user = await User.create(
        username="zhangsan",
        password="zhangsan123",
        age=20
    )
    print(f"创建用户成功：id={user.id}, username={user.username}")

    # 批量创建用户
    users = await User.bulk_create([
        User(username="lisi", password="lisi123", age=22),
        User(username="wangwu", password="wangwu123", age=25)
    ])
    print(f"批量创建用户成功，共创建{len(users)}个用户")

if __name__ == "__main__":
    run_async(create_user())

代码说明：

create()：创建单个数据对象，返回创建后的模型实例。
bulk_create()：批量创建数据，接收模型实例列表，效率高于多次调用create()。

方法2：先实例化再调用`save()`

async def create_user_by_save():
    await init_db()
    # 实例化模型
    user = User(username="zhaoliu", password="zhaoliu123", age=18)
    # 保存到数据库
    await user.save()
    print(f"保存用户成功：id={user.id}, username={user.username}")

if __name__ == "__main__":
    run_async(create_user_by_save())

代码说明：适用于需要先对实例进行其他操作，再保存到数据库的场景。

4.2 数据读取（Read）

tortoise-orm提供了丰富的查询方法，支持过滤、排序、分页等操作，常用方法包括all()、filter()、get()、first()等。

async def query_user():
    await init_db()

    # 1. 查询所有用户
    all_users = await User.all()
    print("所有用户：")
    for user in all_users:
        print(f"id={user.id}, username={user.username}, age={user.age}")

    # 2. 过滤查询：查询年龄大于20的用户
    filter_users = await User.filter(age__gt=20).all()
    print("\n年龄大于20的用户：")
    for user in filter_users:
        print(f"username={user.username}, age={user.age}")

    # 3. 精确查询：根据用户名查询用户（get()方法，查询不到会抛异常）
    try:
        user = await User.get(username="zhangsan")
        print(f"\n精确查询用户：id={user.id}, age={user.age}")
    except User.DoesNotExist:
        print("用户不存在")

    # 4. 排序查询：按年龄降序排列
    order_users = await User.all().order_by("-age")
    print("\n按年龄降序排列的用户：")
    for user in order_users:
        print(f"username={user.username}, age={user.age}")

    # 5. 分页查询：获取第2页数据，每页2条
    page_users = await User.all().offset(2).limit(2)
    print("\n分页查询结果：")
    for user in page_users:
        print(f"username={user.username}, age={user.age}")

if __name__ == "__main__":
    run_async(query_user())

代码说明：

filter()：支持多种查询条件，如age__gt=20（年龄大于20）、age__lt=30（年龄小于30）、username__contains="zhang"（用户名包含zhang）。
get()：查询单个对象，查询结果不存在会抛出DoesNotExist异常，存在多个会抛出MultipleObjectsReturned异常。
order_by()：排序，字段前加-表示降序。
offset()：跳过指定数量的数据，limit()：限制返回数据的数量，两者结合实现分页。

4.3 数据更新（Update）

更新数据有两种方式：模型实例更新和批量更新。

方式1：模型实例更新

async def update_user():
    await init_db()
    # 查询要更新的用户
    user = await User.get(username="zhangsan")
    # 修改属性
    user.age = 21
    user.password = "new_zhangsan123"
    # 保存更新
    await user.save()
    print(f"更新用户成功：username={user.username}, 新年龄={user.age}")

if __name__ == "__main__":
    run_async(update_user())

方式2：批量更新

async def bulk_update_user():
    await init_db()
    # 批量更新年龄小于20的用户，将is_active设为False
    update_count = await User.filter(age__lt=20).update(is_active=False)
    print(f"批量更新成功，共更新{update_count}个用户")

if __name__ == "__main__":
    run_async(bulk_update_user())

代码说明：update()方法返回受影响的行数，适用于批量修改数据，效率更高。

4.4 数据删除（Delete）

删除数据同样支持单个删除和批量删除。

async def delete_user():
    await init_db()
    # 1. 单个删除：查询后删除
    user = await User.get(username="zhaoliu")
    await user.delete()
    print(f"删除用户成功：username={user.username}")

    # 2. 批量删除：删除is_active为False的用户
    delete_count = await User.filter(is_active=False).delete()
    print(f"批量删除成功，共删除{delete_count}个用户")

if __name__ == "__main__":
    run_async(delete_user())

五、外键关联与多表查询

tortoise-orm支持外键关联，实现多表之间的关联查询，以User和Article模型为例（一个用户可以发布多篇文章）。

5.1 定义关联模型

class Article(Model):
    id = fields.IntField(pk=True)
    title = fields.CharField(max_length=100, description="文章标题")
    content = fields.TextField(description="文章内容")
    # 外键关联User模型，on_delete=fields.CASCADE表示删除用户时同时删除文章
    author = fields.ForeignKeyField("models.User", related_name="articles", on_delete=fields.CASCADE)
    created_at = fields.DatetimeField(auto_now_add=True)

    class Meta:
        table = "article"

代码说明：

ForeignKeyField：定义外键，第一个参数为关联的模型（格式为模块名.模型名）。
related_name：反向关联名称，通过User.articles可查询用户发布的所有文章。
on_delete：外键删除策略，fields.CASCADE为级联删除，fields.SET_NULL为设为NULL（需字段允许null=True）。

5.2 关联查询实战

async def relation_query():
    await init_db()
    # 1. 创建用户并关联文章
    user = await User.create(username="author1", password="author123", age=30)
    await Article.bulk_create([
        Article(title="tortoise-orm入门", content="tortoise-orm是一款异步ORM工具", author=user),
        Article(title="异步编程实战", content="Python异步编程技巧", author=user)
    ])

    # 2. 正向查询：查询文章的作者信息
    article = await Article.get(title="tortoise-orm入门")
    # 预加载作者信息，避免N+1查询问题
    await article.fetch_related("author")
    print(f"文章标题：{article.title}，作者：{article.author.username}")

    # 3. 反向查询：查询用户发布的所有文章
    user = await User.get(username="author1")
    articles = await user.articles.all()
    print(f"\n用户{user.username}发布的文章：")
    for art in articles:
        print(f"标题：{art.title}")

if __name__ == "__main__":
    run_async(relation_query())

代码说明：

fetch_related()：预加载关联数据，解决ORM中的N+1查询性能问题。
反向关联：通过related_name（如articles）直接查询关联数据，语法简洁。

六、数据库迁移

在实际开发中，模型结构会不断变化，tortoise-orm提供了aerich工具来管理数据库迁移，类似于Django的makemigrations和migrate。

6.1 安装aerich

pip install aerich

6.2 初始化迁移配置

创建配置文件pyproject.toml（或在项目根目录执行命令生成）

aerich init -t tortoise_config.TORTOISE_ORM

初始化数据库

aerich init-db

6.3 生成迁移文件与执行迁移

当模型修改后，生成迁移文件：

aerich migrate --name update_user_model

执行迁移，更新数据库表结构：

aerich upgrade

七、实际项目案例：异步用户管理系统

下面通过一个简单的异步用户管理系统，整合tortoise-orm的核心功能，实现用户的注册、查询、更新和删除。

7.1 项目目录结构

user_manage/
├── main.py          # 主程序入口
├── models.py        # 模型定义
└── requirements.txt # 依赖包列表

7.2 编写模型文件`models.py`

from tortoise import fields
from tortoise.models import Model

class User(Model):
    id = fields.IntField(pk=True)
    username = fields.CharField(max_length=30, unique=True, description="用户名")
    password = fields.CharField(max_length=100, description="密码")
    age = fields.IntField(default=0, description="年龄")
    is_active = fields.BooleanField(default=True, description="是否激活")
    created_at = fields.DatetimeField(auto_now_add=True, description="创建时间")
    updated_at = fields.DatetimeField(auto_now=True, description="更新时间")

    class Meta:
        table = "user"
        indexes = [("username",)]

7.3 编写主程序`main.py`

import asyncio
from tortoise import Tortoise, run_async
from models import User

# 数据库配置
TORTOISE_ORM = {
    "connections": {"default": "sqlite://user_manage.db"},
    "apps": {
        "models": {
            "models": ["models"],
            "default_connection": "default",
        },
    },
}

# 初始化数据库
async def init_db():
    await Tortoise.init(config=TORTOISE_ORM)
    await Tortoise.generate_schemas()

# 用户注册
async def user_register(username: str, password: str, age: int):
    await init_db()
    try:
        user = await User.create(username=username, password=password, age=age)
        return {"code": 200, "msg": "注册成功", "data": {"user_id": user.id, "username": user.username}}
    except Exception as e:
        return {"code": 500, "msg": f"注册失败：{str(e)}"}

# 查询用户信息
async def user_query(username: str = None):
    await init_db()
    if username:
        try:
            user = await User.get(username=username)
            data = {
                "user_id": user.id,
                "username": user.username,
                "age": user.age,
                "is_active": user.is_active,
                "created_at": user.created_at.strftime("%Y-%m-%d %H:%M:%S")
            }
            return {"code": 200, "msg": "查询成功", "data": data}
        except User.DoesNotExist:
            return {"code": 404, "msg": "用户不存在"}
    else:
        users = await User.all()
        data = []
        for user in users:
            data.append({
                "user_id": user.id,
                "username": user.username,
                "age": user.age,
                "is_active": user.is_active
            })
        return {"code": 200, "msg": "查询成功", "data": data}

# 主函数
async def main():
    # 注册用户
    register_res = await user_register("test_user", "test123", 25)
    print(register_res)

    # 查询单个用户
    query_res = await user_query("test_user")
    print(query_res)

    # 查询所有用户
    all_users_res = await user_query()
    print(all_users_res)

if __name__ == "__main__":
    run_async(main())

7.4 运行项目

执行main.py，输出如下：

{'code': 200, 'msg': '注册成功', 'data': {'user_id': 1, 'username': 'test_user'}}
{'code': 200, 'msg': '查询成功', 'data': {'user_id': 1, 'username': 'test_user', 'age': 25, 'is_active': True, 'created_at': '2024-05-20 15:30:00'}}
{'code': 200, 'msg': '查询成功', 'data': [{'user_id': 1, 'username': 'test_user', 'age': 25, 'is_active': True}]}

八、相关资源

Pypi地址：https://pypi.org/project/tortoise-orm
Github地址：https://github.com/tortoise/tortoise-orm
官方文档地址：https://tortoise-orm.readthedocs.io/en/latest/

关注我，每天分享一个实用的Python自动化工具。

实用工具

Python实用工具：tortoise-orm入门到实战教程

一、tortoise-orm安装与环境配置

1.1 安装tortoise-orm

1.2 验证安装

二、tortoise-orm核心概念与初始化

2.1 核心概念

2.2 数据库初始化

三、tortoise-orm模型定义与字段类型

3.1 模型定义规则

3.2 常用字段类型

四、tortoise-orm核心操作：CRUD实战

4.1 数据创建（Create）

方法1：使用`create()`直接创建

方法2：先实例化再调用`save()`

4.2 数据读取（Read）

4.3 数据更新（Update）

方式1：模型实例更新

方式2：批量更新

4.4 数据删除（Delete）

五、外键关联与多表查询

5.1 定义关联模型

5.2 关联查询实战

六、数据库迁移

6.1 安装aerich

6.2 初始化迁移配置

6.3 生成迁移文件与执行迁移

七、实际项目案例：异步用户管理系统

7.1 项目目录结构

7.2 编写模型文件`models.py`

7.3 编写主程序`main.py`

7.4 运行项目

八、相关资源

更多文章

Python实用工具Upgini：零代码特征工程与数据集增强教程

Python实用工具库excalibur：PDF表格提取与数据处理实战教程

Python实用工具：rows库快速入门与实战指南

Python实用工具：Camelot库——轻松提取PDF表格数据的完整指南

Python实用工具：tortoise-orm入门到实战教程

一、tortoise-orm安装与环境配置

1.1 安装tortoise-orm

1.2 验证安装

二、tortoise-orm核心概念与初始化

2.1 核心概念

2.2 数据库初始化

三、tortoise-orm模型定义与字段类型

3.1 模型定义规则

3.2 常用字段类型

四、tortoise-orm核心操作：CRUD实战

4.1 数据创建（Create）

方法1：使用create()直接创建

方法2：先实例化再调用save()

4.2 数据读取（Read）

4.3 数据更新（Update）

方式1：模型实例更新

方式2：批量更新

4.4 数据删除（Delete）

五、外键关联与多表查询

5.1 定义关联模型

5.2 关联查询实战

六、数据库迁移

6.1 安装aerich

6.2 初始化迁移配置

6.3 生成迁移文件与执行迁移

七、实际项目案例：异步用户管理系统

7.1 项目目录结构

7.2 编写模型文件models.py

7.3 编写主程序main.py

7.4 运行项目

八、相关资源

更多文章

Python实用工具Upgini：零代码特征工程与数据集增强教程

Python实用工具库excalibur：PDF表格提取与数据处理实战教程

Python实用工具：rows库快速入门与实战指南

Python实用工具：Camelot库——轻松提取PDF表格数据的完整指南

方法1：使用`create()`直接创建

方法2：先实例化再调用`save()`

7.2 编写模型文件`models.py`

7.3 编写主程序`main.py`