后端（FastAPI）学习笔记（CLASS 3）：Tortoise ORM

Tortoise ORM是一个易于使用的ORM（对象关系映射器），灵感来自于Django

python有许多现有和成熟的ORM，不幸的是，它们的设计方式与I/O处理方式相反，是相对较新的技术，具有不同的并发模型，最大的变化是关于I/O的处理方式

一般为：数据表对应模型类，字段对应类属性，字段约束对应类的属性值，表记录对应模型类的实例

一、模型定义

1、基本使用

在tortoise ORM中，定义模型类似于Django ORM。我们定义一个模型类，并继承自Model类。下面是一个简单的模型定义实例，包括字段类型和关系

1、安装tortoise ORM：

pip install tortoise-rom

pip install tortoise-orm[asyncmy]

2、配置tortoise ORM

在项目入口文件中配置Tortoise ORM，例如在main.py中：

from tortoise import Tortoise, run_async

async def init():
    await Tortoise.init(
        db_url='mysql://root:mysql@localhost:3306/library-management',
        modules={'models': ['__main__']}
    )
    
    await Tortoise.generate_schemas()
    
run_async(init())

3、定义模型

假设我们有User和Post模型，其中User和Post之间是一对多的关系

2、字段类型

Tortoise ORM 提供了多种字段类型，以下是一些常用的字段类型：

3、关系类型

一对多（ForeignKeyField）

多对多（ManyToManyField）

一对一（OneToOneField）

1、一对多关系

一个用户可以有多个帖子：

# 定义一个用户表（用户模型）
# 1个用户 ---> 可以发布多篇文章
class UserModel(Model):
    id = fields.IntField(primary_key=True, description="用户id")
    nickname = fields.CharField(max_length=255, description="用户昵称")
    username = fields.CharField(max_length=255, description="用户名")
    password = fields.CharField(max_length=255, description="用户密码")
    # 或者
    # posts = fields.ReverseRelation["PostModel"]
    
    def __str__(self):
        return self.nickname
    
    class Meta:
        table = "user"
        table_description = "用户表"
        
class PostModel(Model):
    """帖子表"""
    id = fields.IntField(primary_key=True, description="用户id")
    title = fields.CharField(max_length=255, description="帖子标题")
    content = fields.CharField(description="帖子内容")
    user = fields.ForeignKeyField("models.UserModel", related_name="posts", description="用户id")
    
    def __str__(self):
        return self.title
    
    class Meta:
        table = "Post"
        table_description="帖子表"

2、一对一的关系

一个用户对应一个用户详情

# 定义一个用户详细信息表
# 1个用户 ---> 1个用户信息
class UserInfoModel(Model):
    id = fields.IntField(primary_key = True, description = "用户信息id")
    age = fields.IntField(desciption="年龄")
    sex = fields.CharField(max_length=255, description="性别")
    user = fields.OneToOneField("models.UserModel", related_name="user_info", description="用户id")
    
    def __str__(self):
        return self.user.nickname

3、多对多的关系

多个用户对应多个社群

# 定义一个社群表
# 1个用户可以加入多个社群
# 1个社群有多个用户

class CommunityModel(Model):
    id = fields.IntField(primary_key=True, description="社群id")
    name = fields.CharField(max_length=255, description="社群名称")
    user = fields.ManyToManyField("models.UserModel", related_name="communities", description="用户id")

二、基础查询语法

当然，以下是Tortoise ORM中关联模型查询的详细语法，包括一对多、多对一、多对多等情况的示例

1、模型定义

首先，我们定义几个示例模型

如上代码

2、以下为整体附带创建数据的方法：

from tortoise import Tortoise, Model, fields, run_async

"""
模型类型的设计：
    1、定位继承tortoise.models.Model类
    2、在模型类中定义字段
        字段的属性是tortoise.fields中的字段类型
"""

# 定义一个用户表（用户模型）
# 1个用户 ---> 可以发布多篇文章
class UserModel(Model):
    id = fields.IntField(primary_key=True, description="用户id")
    nickname = fields.CharField(max_length=255, description="用户昵称")
    username = fields.CharField(max_length=255, description="用户名")
    password = fields.CharField(max_length=255, description="用户密码")
    # 或者
    # posts = fields.ReverseRelation["PostModel"]
    
    def __str__(self):
        return self.nickname
    
    class Meta:
        table = "user"
        table_description = "用户表"
        
class PostModel(Model):
    """帖子表"""
    id = fields.IntField(primary_key=True, description="用户id")
    title = fields.CharField(max_length=255, description="帖子标题")
    content = fields.CharField(max_length=255, description="帖子内容")
    user = fields.ForeignKeyField("models.UserModel", related_name="posts", description="用户id")
    
    def __str__(self):
        return self.title
    
    class Meta:
        table = "Post"
        table_description="帖子表"
        
        
# 定义一个用户详细信息表
# 1个用户 ---> 1个用户信息
class UserInfoModel(Model):
    id = fields.IntField(primary_key = True, description = "用户信息id")
    age = fields.IntField(description="年龄")
    sex = fields.CharField(max_length=255, description="性别")
    user = fields.OneToOneField("models.UserModel", related_name="user_info", description="用户id")
    
    def __str__(self):
        return self.user.nickname
    

# 定义一个社群表
# 1个用户可以加入多个社群
# 1个社群有多个用户

class CommunityModel(Model):
    id = fields.IntField(primary_key=True, description="社群id")
    name = fields.CharField(max_length=255, description="社群名称")
    user = fields.ManyToManyField("models.UserModel", related_name="communities", description="用户id")
    
    
async def create_data():
    # 创建数据:模型类.create()
    user = await UserModel.create(nickname="张三", username="zhangsan", password="123456")
    userinfo = await UserInfoModel.create(age=18, sex="男", user=user)
    print(user, userinfo)
    
async def init():
    # 初始化连接数据库配置
    await Tortoise.init(
        db_url='mysql://root:20050824@localhost:3306/test',
        modules={'models': ['__main__']}
    )
    await create_data()
    # 在数据库中创建表
    # await Tortoise.generate_schemas()
    
if __name__ == '__main__':
    run_async(init())

3、以下为查询语法：

async def find_all_user():
    # all: 获取所有数据
    users = await UserModel.all()
    for user in users:
        print(user)
    # filter: 过滤所有数据
    users = await UserModel.filter(id=1)
    for user in users:
        print(user)
    # get: 获取单条数据,数据不存在会报错
    user = await UserModel.get(id=1)
    print(user.__dict__['nickname'])
    # get_or_none: 获取单条数据,数据不存在则返回空

4、以下为更新语法：

async def update_user():
    # 方式一: 根据对象.update_from_dict
    user = await UserModel.get(id=1)
    user = await user.update_from_dict({
        "nickname": "张三2",
        "username": "zhangsan",
        "password": "123456"
    })
    # 修改完要用save函数保存
    
    # 方式二: 根据对象.属性=值
    user = await UserModel.get(id=1)
    user.nickname = '老张呀'
    
    await user.save()
    print(user.__dict__)
    
    # 查询集的update方法
    await UserModel.filter(id=1).update(nickname="老张")

5、以下为删除语法：

async def delete_user():
    # 方式一: 使用查询集的delete方法, 进行批量删除
    await UserModel.filter(id=1).delete()
    # 方式二: 使用模型对象的delete方法,进行单个删除
    user = await UserModel.get(id=1)
    await user.delete()

三、关联查询

1、一对一关联查询

从User获取UserInfo的信息

# 一对一关联查询的使用
async def one_to_one_query():
    user = await UserModel.get(id = 1).prefetch_related('user_info')
    if user and hasattr(user, "user_info"):
        print(user.user_info.address)

2、一对多关联查询

从User获取Post

async def get_user_posts(user_id: int):
    user = await UserModel.get(id=user_id).prefetch_related('posts')
    posts = await user.posts.all().values("id", "title", "content")
    print(f"User: {user.nickname}, Posts: {posts}")
    

3、多对一关联查询

通过Post获取User

# 多对一关联查询的使用
async def many_to_one_query():
    post = await PostModel.get(id=1).prefetch_related("user")
    if post and hasattr(post, "user"):
        result = {
            "id": post.id,
            "title": post.title,
            "content": post.content,
            "user": {
                "id": post.user.id,
                "nickname": post.user.nickname
            }
        }
        print(result)

4、多对多关联查询

从User中获取Community

# 多对多的查询
async def many_to_mant_query():
    user = await UserModel.get(id=1).prefetch_related("communities")
    if user and hasattr(user, "communities"):
        communities = []
        for community in user.communities:
            communities.append({
                "id": community.id,
                "name": community.name
            })
        result = {
            "id": user.id,
            "nickname": user.nickname,
            "communities": communities
        }
        print(result)

反过来也是一样的

四、高级查询

1、排序

获取特定用户及其所有帖子，并根据帖子标题排序

# 排序查询
async def sort_user_all():
    users = await UserModel.all().order_by("id").values("id", "nickname", "username")
    print(users)

2、过滤查询

获取某个过滤条件下所需的信息

# 排序查询
async def sort_user_all():
    users = await UserModel.filter(id__gt=1).values("id", "nickname", "username")
    print(users)

gt（大于）、lt（小于）、gte（大于等于）、lte（小于等于）

icontains（包含）

3、分页查询

获取用户的所有帖子，并进行分页（每页显示5条记录）：

# 分页查询
    """
    offset方法：指定数据获取的起始位置
    limit方法：指定每页返回多少条数据
    """
async def limit_page_user_all(page, size):
    user = await UserModel.all().offset((page - 1) * size ).limit(size)
    print(user)

4、聚合查询

在查询的信息中返回总数，平均数等信息

from tortoise.functions import Count, Avg, Max, Min

# 聚合查询
async def annotate_query():
    user = await UserModel.annotate(
        post_count = Count("posts")
    ).all()

5、自定义sql查询

有时候你可能需要执行原始SQL查询，Tortoise ORM也提供了支持。使用了Tortoise.get_connection().execute_query_dict()方法执行原始SQL查询，并获取结果

async def execute_raw_sql() -> None:
    # 获取数据库连接对象
    db = Tortoise.get_connection('defalut')
    # 执行查询语句并返回字典
    users = await db.execute_query_dict("select * from user")
    print(users)

* 总结

1、按条件查询：使用filter方法添加查询条件

2、排序：使用order_by方法对查询结果进行排序

3、分页：使用offset和limit方法进行分页查询

4、关联查询：使用prefetch_related或select_related方法进行关联查询

5、聚合查询：使用annotate进行联合查询

五、数据库迁移

1、generate_schemas

tortoise ORM自带的generate_schemas方法进行迁移

2、使用Aerich

tortoise ORM自带的generate_schemas方法是为了快速开发时使用的，它会在数据库中创建所有必要的表结构，但不使用于生产环境的数据迁移。为了进行更复杂的数据库迁移，推荐使用第三方工具Alembic

安装Aerich：

pip install aerich

迁移配置：

# 关于tortoise的配置
TORTOISE_ORM = {
    'connections': {
        'default': {
            'engine': 'tortoise.backends.mysql',
            'credentials': {
                'host': 'localhost',
                'port': '3306',
                'user': 'root',
                'password': '20050824',
                'database': 'demo'
            }
        }
    },
    'apps': {
        'models': {
            'modesl': ['aerich.models', '4.6transform'],
            'default_connection': 'default'
        }
    }
}

# 注册ORM模型
register_tortoise(app, config=TORTOISE_ORM)

迁移命令：

在项目根目录运行一下命令：

aerich init -t main.TORTOISE_ORM

其中app.TORTOISE_ORM是Tortoise ORM的配置路径。例如，如果Tortoise配置在main.py中，配置路径可以是main.TORTOISE_ORM

生成迁移文件，初始化数据库：

aerich init -db

每次修改模型后，运行以下命令生成迁移文件：

aerich migrate

应用迁移：

aerich upgrade