目录
一.ORM生态
二.Tortoise-ORM配置
三.Aerich 迁移工具的使用
1.Aerich命令
2.Aerich迁移
四.Tortoise-ORM 的模型定义
五.Tortoise-ORM CRUD增删改查
六.Tortoise-ORM 的关联关系建立
1.一对一
2.一对多
3.多对多
七.Tortoise-ORM 的关联关系数据操作
一.ORM生态
ORM介绍
ORM（Object-Relational Mapping，对象关系映射）是一种编程技术，用于在面向对象编程语言和关系型数据库之间建立映射。它允许开发者通过操作对象的方式来与数据库进行交互，而无需直接编写复杂的 SQL语句。主要特点包括：
对象与数据库表的映射：ORM将数据库中的表映射为编程语言中的类，每一行数据对应一个对
象，表的列对应对象的属性。
简化数据库操作：开发者可以使用面向对象的方法（如创建、查询、更新、删除对象）来操作数据
库，而无需手动编写SQL。
2
跨数据库兼容：ORM通常支持多种数据库（如MySQL、PostgreSQL、SQLite），通过统一的接
口减少数据库切换的成本。
3
提高开发效率：通过自动化SQL生成和查询优化，减少重复代码，提升开发速度。
工作原理
ORM框架定义了一个映射层，将类和数据库表关联起来
开发者通过ORM的API操作对象，ORM自动将操作翻译成SQL语句，执行数据库操作并返回结
果
优点
提高代码可读性和维护性
减少直接SQL操作带来的错误。
支持复杂查询和关系（如一对多、多对多）
缺点
性能可能略低于原生SQL（因抽象层开销）
对于非常复杂的查询，可能需要直接编写SQL
ORM工具介绍
SQLAlchemy（同步/异步）：≈80%企业项目首选，功能完备、社区成熟，支持复杂查询和事务
管理。
Tortoise（异步）：语法类似Django ORM，适合异步优先项目，集成简便
GINO（异步）：轻量级，基于SQLAlchemy Core的异步扩展，适合高性能API
选型建议
传统企业项目→SQLAlchemy（成熟稳定）
全异步微服务→Tortoise-ORM或GINO
二.Tortoise-ORM配置
为什么选择Tortoise-ORM
异步支持，与FastAPI无缝集成
简单易用的模型定义，类似Django ORM
支持复杂关系（一对一、一对多、多对多）
自动生成表结构，适合快速开发
环境配置
安装必要的依赖：
uv pip installtortoise-ormaerichaiomysqltomlkit
# ORM基础配置 from tortoise.contrib.fastapi import register_tortoise from typing import Dict from fastapi import FastAPI app = FastAPI() # Tortoise-ORM 配置 TORTOISE_ORM: Dict = { # connections：定义数据库连接字符串 "connections": { # 开发环境使用 SQLite（基于文件，无需服务器） # "default": "sqlite://db.sqlite3", # 生产环境示例：PostgreSQL # "default": "postgres://user:password@localhost:5432/dbname", # 生产环境示例：MySQL "default": "mysql://root:123456@192.168.31.152:3306/fastapi_db", }, # 定义应用模块，models 列表包含模型文件路径和Aerich的迁移模型 "apps": { "models": { "models": ["model19", "aerich.models"], # 模型模块和 Aerich 迁移模型 "default_connection": "default", #这里的default指的就是connections中的default } }, # 连接池配置（推荐） "use_tz": False, # 是否使用时区 "timezone": "UTC", # 默认时区 "db_pool": { "max_size": 10, # 最大连接数 "min_size": 1, # 最小连接数 "idle_timeout": 30 # 空闲连接超时（秒） } } # 注册 Tortoise-ORM 到 FastAPI 应用 register_tortoise(app, config=TORTOISE_ORM, generate_schemas=True, # 开发环境自动生成表结构 add_exception_handlers=True # 添加默认异常处理 ) if __name__ == '__main__': import uvicorn uvicorn.run('main19:app', host='127.0.0.1', port=8000, reload=True)三.Aerich 迁移工具的使用
1.Aerich命令
Aerich是Tortoise-ORM的数据库迁移工具，用于管理数据库结构的变更。
Aerich初始化
在项目根目录运行以下命令：
aerich init-tmain.TORTOISE_ORM
这将生成：
pyproject.toml：Aerich配置文件，指定迁移配置。
migrations/：迁移文件目录，存放生成的.sql文件。
aerich init-db
生成和应用迁移
1
生成迁移文件： 当模型发生变更时，运行以下命令生成迁移文件：
aerich migrate--name"info"
2
应用迁移： 运行以下命令将迁移应用到数据库：
aerich upgrade
3
验证迁移： 检查迁移历史
aerich history
4
回滚迁移：回退到指定版本
aerich downgrade
2.Aerich迁移
先安装database client插件
填写好信息，点连接
成功
点加号
在新连接这里点加号，创建数据库，点击Run
新建数据库成功
创建一个类文件model19.py
from tortoise.models import Model from tortoise.fields import CharField, DatetimeField, BooleanField class User(Model): id = CharField(max_length=36, pk=True) # 主键，UUID 字符串 username = CharField(max_length=50, unique=True) # 用户名，唯一 email = CharField(max_length=255, unique=True) # 邮箱，唯一 is_active = BooleanField(default=True) # 是否激活 email666 = CharField(max_length=255, unique=True) # 邮箱，唯一 class Meta: table = "users" # 自定义表名 ordering = ["-created_at"] # 默认按创建时间降序排序 def __str__(self): return self.username ''' # 根据main19.py文件中的User模型去生成迁移文件,因为此文件里的TORTOISE_ORM变量相当于配置文件 # 这两个init初始化命令一般在项目周期里只执行一次 aerich init -t main19.TORTOISE_ORM # 初始化 aerich init-db # 创建迁移脚本，一般是生成一堆类对应的迁移脚本 # 这两个迁移命令一般在项目周期一般执行很多次，每次执行迁移(migrate)后都需要同步迁移（upgrade）到数据库 aerich migrate --name "注释" # 迁移 aerich upgrade # 同步迁移到数据库 '''配置和注册文件main19.py
# ORM基础配置 from tortoise.contrib.fastapi import register_tortoise from typing import Dict from fastapi import FastAPI app = FastAPI() # Tortoise-ORM 配置 TORTOISE_ORM: Dict = { # connections：定义数据库连接字符串 "connections": { # 开发环境使用 SQLite（基于文件，无需服务器） # "default": "sqlite://db.sqlite3", # 生产环境示例：PostgreSQL # "default": "postgres://user:password@localhost:5432/dbname", # 生产环境示例：MySQL "default": "mysql://root:root@127.0.0.1:3306/fastapi_db", }, # 定义应用模块，models 列表包含模型文件路径和Aerich的迁移模型 "apps": { "models": { # 根据model19.py文件中的User模型去生成迁移文件 "models": ["model19", "aerich.models"], # 模型模块和 Aerich 迁移模型 "default_connection": "default", #这里的default指的就是connections中的default } }, # 连接池配置（推荐） "use_tz": False, # 是否使用时区 "timezone": "UTC", # 默认时区 "db_pool": { "max_size": 10, # 最大连接数 "min_size": 1, # 最小连接数 "idle_timeout": 30 # 空闲连接超时（秒） } } # 注册 Tortoise-ORM 到 FastAPI 应用 register_tortoise(app, config=TORTOISE_ORM, generate_schemas=True, # 开发环境自动生成表结构 add_exception_handlers=True # 添加默认异常处理 ) if __name__ == '__main__': import uvicorn uvicorn.run('main19:app', host='127.0.0.1', port=8000, reload=True)依次执行那四个命令即可，model19.py文件里注释写的非常明白。
后续如果需要修改表结构，直接修改代码即可，比如把User类里的某个成员变量删了，然后执行迁移和同步，数据库就随之改变了，确实好用啊，太方便了。
aerich migrate --name "注释" # 迁移 aerich upgrade # 同步迁移到数据库四.Tortoise-ORM 的模型定义
Tortoise-ORM的模型通过继承tortoise.models.Model类定义，每个模型对应数据库中的一张表。
字段定义使用Tortoise-ORM提供的字段类型，结合字段参数来指定约束和行为。
继承Model：所有模型必须继承tortoise.models.Model。
字段定义：每个字段对应数据库表中的一列，使用Tortoise-ORM的字段类（如CharField、
BooleanField）。
Meta类：用于定义模型的元数据，如表名、排序规则等。
核心字段类型详解
Tortoise-ORM提供了多种字段类型，用于定义数据库列的类型和行为。以下是常用的字段类型及其典型用途：
字段参数详解
字段支持多种参数，用于定义约束、默认值和其他行为。以下是常用参数：
max_length：字符串字段的最大长度（CharField必填）
null：是否允许字段为空（null=True表示数据库允许NULL）
default：字段默认值（如default=0、default=True）
unique：是否唯一（unique=True确保字段值在表中唯一）
index：是否创建索引（index=True提高查询性能）
description：字段描述（用于文档或数据库注释）
pk：是否为主键（pk=True表示该字段是主键）
validators：自定义验证函数
模型元数据（Meta类）
Meta类用于定义模型的元数据，控制表结构和查询行为。常用属性包括：
table：自定义数据库表名（如table="users"）
unique_together：定义联合唯一约束（如unique_together=[("field1", "field2")]）
indexes：定义索引（如indexes=[("field1", "field2")]）
ordering：默认排序规则（如ordering=["-created_at", "username"]）
from tortoise.models import Model from tortoise.fields import CharField, TextField, IntField, BooleanField, DatetimeField class User(Model): id = CharField(max_length=36, pk=True) name = CharField(max_length=64, unique=True) email = CharField(max_length=255) is_active = BooleanField(default=True) age = IntField(default=0) # 创建时间 created_at = DatetimeField(auto_now_add=True) class Meta: table = 't_user' # table/view unique_together = ('name', 'email') # 联合唯一 ordering = ['-created_at'] # -降序 默认升序 # 创建索引 indexes = [ 'email', ] ''' aerich init -t main20.TORTOISE_ORM # 初始化 aerich init-db # 创建迁移脚本 aerich migrate --name "注释" # 迁移 aerich upgrade # 同步迁移 '''五.Tortoise-ORM CRUD增删改查
model21.py 学生类
from tortoise import fields, models class Student(models.Model): """ 学生模型，包含基本信息。 用于演示单表的增删改查操作。 """ id = fields.IntField(pk=True, description="学生ID，主键") name = fields.CharField(max_length=50, description="学生姓名") age = fields.IntField(null=True, description="学生年龄，可为空") email = fields.CharField(max_length=100, unique=True, null=True, description="学生邮箱，唯一") class Meta: table = "students" def __str__(self): return f"Student: {self.name}, Age: {self.age}, Email: {self.email}"main21.py 接口方式实现CRUD
# ORM基础配置 from crud21 import create_student from tortoise.contrib.fastapi import register_tortoise from typing import Dict from fastapi import FastAPI app = FastAPI() # Tortoise-ORM 配置 TORTOISE_ORM: Dict = { "connections": { # 开发环境使用 SQLite（基于文件，无需服务器） # "default": "sqlite://db.sqlite3", # 生产环境示例：PostgreSQL # "default": "postgres://user:password@localhost:5432/dbname", # 生产环境示例：MySQL "default": "mysql://root:123456@192.168.31.152:3306/fastapi_db3", }, "apps": { "models": { "models": ["model21", "aerich.models"], # 模型模块和 Aerich 迁移模型 "default_connection": "default", } }, # 连接池配置（推荐） "use_tz": False, # 是否使用时区 "timezone": "UTC", # 默认时区 "db_pool": { "max_size": 10, # 最大连接数 "min_size": 1, # 最小连接数 "idle_timeout": 30 # 空闲连接超时（秒） } } register_tortoise(app, config=TORTOISE_ORM, generate_schemas=True, # 开发环境自动生成表结构 add_exception_handlers=True # 添加默认异常处理 ) #创建对象时自动在数据库中插入了 @app.get('/create') async def create(): stu = await create_student('孙权', 18, 'sunquan@163.com') return stu # 生成一个传递参数创建学生数据的路由 @app.post('/create2') async def create2(name: str, age: int = None, email: str = None): stu = await create_student(name, age, email) return stu if __name__ == '__main__': import uvicorn uvicorn.run('main21:app', host='127.0.0.1', port=8000, reload=True)crud21.py 脚本方式实现CRUD
from model21 import Student from tortoise import Tortoise, run_async async def create_student(name: str, age: int = None, email: str = None) -> Student: try: stu = await Student.create(name=name, age=age, email=email) return stu except Exception as e: print("创建失败！") async def update_student(stu_id: int, name: str = None, age: int = None, email: str = None) -> Student: stu = await Student.get(id=stu_id) # 判断用户传递了哪个数据，如果没有传递，就不更新 if name: stu.name = name if age is not None: stu.age = age if email is not None: stu.email = email # 保存 await stu.save() return stu async def delete_student(stu_id: int) -> None: stu = await Student.get(id=stu_id) await stu.delete() # 单条数据 async def get_student(stu_id: int) -> Student: stu = await Student.get(id=stu_id) return stu # 多条数据 async def get_students(name: str) -> list[Student]: ''' 完成匹配 ''' stus = await Student.filter(name=name) return stus async def get_students2(name: str) -> list[Student]: ''' 模糊查询 ''' stus = await Student.filter(name__contains=name) return stus async def get_students3() -> list[Student]: ''' 获取所有数据 ''' stus = await Student.all() return stus # 编写查询名字为曹操的年龄等于20的记录 async def get_students4() -> list[Student]: stus = await Student.filter(name="曹操", age=20) return stus # 1. 直接测试脚本 # 2. 直接应用到接口 async def init(): await Tortoise.init( db_url="mysql://root:123456@192.168.31.152:3306/fastapi_db3", modules={"models": ["model21"]} ) await Tortoise.generate_schemas() async def main(): await init() # stu1 = await create_student(name="曹操") # print(f"创建成功 ID:{stu1.id} Name:{stu1.name} Email:{stu1.email}") # stu2 = await create_student(name="刘备", age=30) # print(f"创建成功 ID:{stu2.id} Name:{stu2.name} Email:{stu2.email}") # stu3 = await create_student(name="张飞", age=30, email="zf@163.com") # print(f"创建成功 ID:{stu3.id} Name:{stu3.name} Email:{stu3.email}") # stu4 = await create_student(name="关羽", age=30, email="zf@163.com") # print(f"创建成功 ID:{stu4.id} Name:{stu4.name} Email:{stu4.email}") # await update_student(stu_id=3, age=36) # await update_student(stu_id=1, email="cc@qq.com", age=40) # await delete_student(stu_id=9) stu = await get_student(3) print(stu) stus1 = await get_students("刘备") print(stus1) stus2 = await get_students2("曹") print(stus2) stus3 = await get_students3() print(stus3) if __name__ == '__main__': run_async(main())六.Tortoise-ORM 的关联关系建立
我们将模拟一个学校管理系统，包含以下实体：
学生（Student）：每个学生有唯一的个人信息档案（1对1）。
成绩（Grade）：一个学生可以有多份成绩记录（1对多）。
课程（Course）：学生和课程之间是多对多关系（通过成绩表关联）。
1.一对一
# 关联关系建立1对1 from tortoise import fields, models, Tortoise, run_async class Student(models.Model): id = fields.IntField(pk=True) name = fields.CharField(max_length=50) # 参数：on_delete=fields.CASCADE 表示当学生被删除时，其个人信息StudentProfile也会被删除 # 参数：related_name="student" 表示在StudentProfile模型中，有一个字段指向Student模型，用于反向查询 profile = fields.OneToOneField( "models.StudentProfile", on_delete=fields.CASCADE, related_name="student") ''' id name profile_id 1 张三 1 2 李四 2 id address phone 1 上海 123456 2 北京 789456 ''' class StudentProfile(models.Model): id = fields.IntField(pk=True) address = fields.CharField(max_length=100) phone = fields.CharField(max_length=20) async def init(): await Tortoise.init( db_url="mysql://root:123456@192.168.31.152:3306/fastapi_db4", modules={"models": ["model22"]} ) await Tortoise.generate_schemas() if __name__ == "__main__": run_async(init())2.一对多
# 关联关系建立1对多 from tortoise import fields, models, Tortoise, run_async class Student(models.Model): id = fields.IntField(pk=True) name = fields.CharField(max_length=50) profile = fields.OneToOneField( "models.StudentProfile", on_delete=fields.CASCADE, related_name="student") ''' 学生表 id name profile_id score_ids score_id1 score_id2 ... 1 张三 1 1-2-3-5-8-6-7 1 2 2 李四 2 成绩表 id score stu_id 1 90 1 2 80 1 3 70 1 4 60 2 5 50 1 ''' class StudentProfile(models.Model): id = fields.IntField(pk=True) address = fields.CharField(max_length=100) phone = fields.CharField(max_length=20) #一对多，在多的一方定义外键 class Grade(models.Model): id = fields.IntField(pk=True) score = fields.FloatField() student = fields.ForeignKeyField( "models.Student", related_name="grades", on_delete=fields.CASCADE) async def init(): await Tortoise.init( db_url="mysql://root:123456@192.168.31.152:3306/fastapi_db4", modules={"models": ["model23"]} ) await Tortoise.generate_schemas() if __name__ == "__main__": run_async(init())3.多对多
model24.py
# 关联关系建立多对多 from tortoise import fields, models, Tortoise, run_async ''' 学生表 id name profile_id 1 张三 1 2 李四 2 成绩表 id score stu_id 1 90 1 2 80 1 3 70 1 4 60 2 5 50 1 课程表 id name 1 python 2 java 3 c++ 学生课程表 stu_id course_id 1 1 1 2 2 3 ''' class Student(models.Model): id = fields.IntField(pk=True) name = fields.CharField(max_length=50) profile = fields.OneToOneField( "models.StudentProfile", on_delete=fields.CASCADE, related_name="student", null=True) class StudentProfile(models.Model): id = fields.IntField(pk=True) address = fields.CharField(max_length=100) phone = fields.CharField(max_length=20) class Grade(models.Model): id = fields.IntField(pk=True) score = fields.FloatField() student = fields.ForeignKeyField( "models.Student", related_name="grades", on_delete=fields.CASCADE) class Course(models.Model): id = fields.IntField(pk=True) name = fields.CharField(max_length=50) # students = fields.ManyToManyField( # "models.Student", related_name="courses", through="studentcourse") class StudentCourse(models.Model): students = fields.ForeignKeyField("models.Student", related_name="courses") courses = fields.ForeignKeyField("models.Course", related_name="students") class Meta: unique_together = ("students", "courses") async def init(): await Tortoise.init( db_url="mysql://root:root@127.0.0.1:3306/fastapi_db4", modules={"models": ["model24"]} ) await Tortoise.generate_schemas() if __name__ == "__main__": run_async(init())七.Tortoise-ORM 的关联关系数据操作
基于关联关系CRUD
# 关联关系的增删改查询 from tortoise import Tortoise, run_async from model24 import Student, Course, StudentCourse, Grade, StudentProfile async def init(): await Tortoise.init( db_url="mysql://root:123456@192.168.31.152:3306/fastapi_db4", modules={"models": ["model24"]} ) async def create_data(): await init() # 1对1 # 学生对象的id会自动生成，profile对象是关联用的，所以不用必须赋初值 # 下面两句create语句会落实到数据库 # stu1 = await Student.create(name="董卓") # pro1 = await StudentProfile.create(address="上海", phone="123456") # 这句代码只是在内存里赋值，并不会落实到数据库，所以需要save操作 # stu1.profile = pro1 # 保存学生对象，会自动保存关联的profile对象，并落实到数据库 # await stu1.save() # pro2 = await StudentProfile.create(address="北京", phone="123456") # stu2 = await Student.create(name="刘备", profile=pro2) # 1对多 stu3 = await Student.create(name="孙权") await Grade.create(score=100, student=stu3) await Grade.create(score=90, student=stu3) await Grade.create(score=80, student=stu3) # 多对多 stu4 = await Student.create(name="关羽") stu5 = await Student.create(name="张飞") cou1 = await Course.create(name="Python") cou2 = await Course.create(name="C++") await StudentCourse.create(students=stu4, courses=cou1) await StudentCourse.create(students=stu4, courses=cou2) await StudentCourse.create(students=stu5, courses=cou1) await StudentCourse.create(students=stu5, courses=cou2) async def update_data(): await init() # 1对1 # 先从数据库取出 stu = await Student.get(id=1) pro = await StudentProfile.create(address="上海", phone="12345678901") stu.profile = pro # 存回数据库 await stu.save() # 1对多 # 方式一 stu2 = await Student.get(id=1) grade1 = await Grade.get(id=1) grade1.student = stu2 await grade1.save() # 方式二 await Grade.filter(id=6).update(student=stu2) async def query_data(): await init() stu = await Student.get(id=1) print(stu.name) # 关联对象profile需要异步获取 方式一 pro = await stu.profile print(pro.address) print(pro.phone) # 关联对象profile需要异步获取 方式二 stu2 = await Student.get(id=1).prefetch_related('profile') # 获取关联数据 print(stu2.profile.address) print(stu2.profile.phone) #all()加不加都行，一般不影响查询结果 stu3 = await Student.get(id=9).prefetch_related('grades').all() for grade in stu3.grades: print(grade.score) async def delete_data(): await init() # grade = await Grade.get(id=9) # await grade.delete() #级联删除 stu = await Student.get(id=9) await stu.delete() if __name__ == '__main__': # run_async(create_data()) # run_async(update_data()) # run_async(query_data()) run_async(delete_data())