Tortoise ORM

官方文档：Tortoise ORM - Tortoise ORM v0.22.0 Documentation

简介

Tortoise ORM：异步，API和Django ORM 大多类似集成Pydantic；多用于asgi starlette / sanic / FastAPI...
Sqlalchemy：支持异步，Flask / Bottle/FastAPI ，集成Pydantic方案可搜索SQLModel

表数据

CREATE TABLE `a` (
   `id` INT(10) NOT NULL AUTO_INCREMENT,
   `data` VARCHAR(500) NOT NULL DEFAULT '0' COLLATE 'utf8mb4_general_ci',
   `create_date` DATETIME NOT NULL DEFAULT CURRENT_TIMESTAMP,
   PRIMARY KEY (`id`) USING BTREE
)
COLLATE='utf8mb4_general_ci'
ENGINE=InnoDB
;

INSERT INTO `a` (`id`, `data`, `create_date`) VALUES
   (1, '23', '2024-11-23 19:58:15'),
   (2, '24', '2024-11-23 19:58:20');

SQLAlchemy + FastAPI

"""
Tortoise ORM & SQLAlchemy
"""# SQlAlchemy
from sqlalchemy import Column, Integer, String, DateTime, func, select
from sqlalchemy.orm import declarative_baseBase = declarative_base()class SqlAlchemyTable(Base):__tablename__ = "a"id = Column(Integer, primary_key=True)data = Column(String)create_date = Column(DateTime, server_default=func.now())from sqlalchemy.ext.asyncio import create_async_engine, AsyncSession
from sqlalchemy.orm import sessionmakerengine = create_async_engine("sqlite+aiosqlite:///basic.sqlite3")
async_session = sessionmaker(engine, expire_on_commit=True, class_=AsyncSession
)def get_db():session = async_session()try:yield sessionfinally:session.close()from fastapi import FastAPI, Depends
from pydantic import BaseModelclass Response(BaseModel):id: intdata: strapp = FastAPI(title="SQLAlchemy")@app.get("/", response_model=Response)
async def index():# 视图里面实例db对象async with async_session() as session:result = await session.execute(select(SqlAlchemyTable).order_by(SqlAlchemyTable.id))return result.scalars().first().__dict__ # 第一条@app.get("/index")
async def index1(db: AsyncSession = Depends(get_db)):result = await db.execute(select(SqlAlchemyTable).order_by(SqlAlchemyTable.id))return result.scalars().all()if __name__ == '__main__':import uvicornuvicorn.run("__main__:app", reload=True)

Tortoise ORM + FastAPI

from tortoise import models, fields
from tortoise.contrib.fastapi import register_tortoise
from tortoise.contrib.pydantic import pydantic_model_creatorclass TortoiseOrm(models.Model):data = fields.CharField(max_length=30)create_date = fields.DatetimeField(auto_now_add=True, null=True)class Meta:table = 'a'Response = pydantic_model_creator(TortoiseOrm, name="OutPut", exclude=("create_date",))from fastapi import FastAPI
app = FastAPI(title="SQLAlchemy & TortoiseORM")register_tortoise(app,db_url="sqlite://basic.sqlite3",modules={"models": ["__main__"]},add_exception_handlers=True,
)@app.get("/", response_model=Response)
async def index():return await TortoiseOrm.all().order_by("id").first()@app.get("/index")
async def index1():return await TortoiseOrm.all().order_by("id")if __name__ == '__main__':import uvicornuvicorn.run("__main__:app", reload=True)

以上是两种orm操作数据库，个人选择了fastapi + tortoise-orm

1、创建模型&基础字段

- ORM

对象关系映射（英语：Object Relational Mapping），表 -> 类; 列 -> 属性

- Tortoise ORM

Tortoise ORM - Tortoise ORM v0.22.0 Documentation

Tortoise ORM 是一个受 Django 启发的易于使用的异步 ORM (对象关系映射器)。

pip install tortoise-orm -i https://pypi.douban.com/simple/

- 此篇文章参考：

https://tortoise.github.io/models.html

Fields - Tortoise ORM v0.22.0 Documentation

表定义

# -*- coding = utf-8 -*-
# @Time : 2024/11/23 20:03
# @Author: Frank
# @File: main.py
# @Software: PyCharm
from enum import IntEnum
from tortoise import models
from tortoise import fieldsclass AbstractModel(models.Model):# 主键，当表里所有属性都没设置pk时，默认生成一个IntField类型 id 的主键id = fields.UUIDField(pk=True)class Meta:# 抽象模型，不生成表abstract = Trueclass MixinTimeFiled:# 添加数据时间created = fields.DatetimeField(null=True, auto_now_add=True)# 修改数据时间modified = fields.DatetimeField(null=True, auto_now=True)class Gender(IntEnum):MAN = 0WOMAN = 1class UserModel(AbstractModel, MixinTimeFiled):# unique 是否唯一 max—length 数据长度 index 是否索引username = fields.CharField(max_length=20, description="描述", unique=True, index=True, source_field="user_name")# null 是否可以为空nickname = fields.CharField(max_length=30, description='nickname', null=True, default='777')# description 字段备注 ddl展示， 此处入库的为 0 or 1gender = fields.IntEnumField(Gender, description='sex', default=Gender.WOMAN)# max——digits 小输点左边最大位数，decimal——places 小数点右边最大位数balance = fields.DecimalField(max_digits=2, decimal_places=2, description='balance')is_admin = fields.BooleanField(default=False)job_info = fields.JSONField(default=dict)class Meta:# 自定义表名，不配置按照类名小写生成table = "test1"table_description = "set table ddl desc"# 多列设置唯一复合所有unique_together = (('gender', 'balance'),)# 排序ordering = ('is_admin',)# 索引indexes = ('balance',)async def init():# Here we create a SQLite DB using file "db.sqlite3"#  also specify the app name of "models"#  which contain models from "app.models"await Tortoise.init(# 数据库连接# db_url='sqlite://db.sqlite3',# 连接mysql pip install aiomysqldb_url='mysql://root:123456@127.0.0.1:3307/python',# 指定管理的models，__main__ 🈯️当前文件的models.Modelmodules={'models': ['__main__']})# Generate the schemaawait Tortoise.generate_schemas()if __name__ == '__main__':from tortoise import Tortoise, run_asyncrun_async(init())

如果需要自定义数据库字段的名称：使用source_field

username = fields.CharField(max_length=20, description="描述", unique=True, index=True, source_field="user_name")

2、单表常用操作

Create

# 单条新增 - 返回 <UserModel: 80f3e3d6-9b1b-4c58-9e77-f33f433184f8>
await UserModel.create(username="777", balance=22.135)# 批量新增 - 返回[<UserModel: 80f3e3d6-9b1b-4c58-9e77-f33f433684f8>, <UserModel: f47c9d3c-155e-4008-9a8e-a58e5a517537>]
await UserModel.bulk_create([UserModel(username=f'{i}', balance=i) for i in range(2)]
)

SQL

INSERT INTO "tableName" 
("id","created","modified","username","nickname","gender","balance","is_admin","job_info")
VALUES 
('a6bf5564-b357-44f5-9132-a1be8f91f47b', '2022-11-13 02:19:35.754746+00:00', '2022-11-13 02:19:35.754759+00:00', '777', '777', 1, '22.14', 0, '{}')# 插入多条
INSERT INTO "tableName" ("id","created","modified","username","nickname","gender","balance","is_admin","job_info") 
VALUES 
('c8556714-5ad8-426b-9fc8-51b179500f81', '2022-11-13 02:19:35.756383+00:00', '2022-11-13 02:19:35.756390+00:00', '0', '777', 1, '0', 0, '{}'),
('0a94335d-3df6-4b4a-8982-408f9799a228', '2022-11-13 02:19:35.756407+00:00', '2022-11-13 02:19:35.756411+00:00', '1', '777', 1, '1', 0, '{}');

Delete

# 返回改动行数
await UserModel.filter(username="777").delete()

SQL

DELETE FROM "tableName" WHERE "username"='777'

Update


# 返回改动行数
await UserModel.filter(username="777").update(gender=0)# 批量更新
users = await UserModel.all()
for user in users:user.balance += 1.68await UserModel.bulk_update(users, fields=['balance'])

SQL

UPDATE "tableName" SET "gender"=0 WHERE "username"='777'# 批量更新
UPDATE "tableName" SET "gender"=0 WHERE "id" IN ('7aeb83f2-c0cb-4d21-a021-6cbe302a2b1e','3a18bbcd-163f-4574-afac-0b9f7ab89652')

Select

all

ps: 分组、排序、分页、聚合也适用 filter

# return [UserModel]
await UserModel.all()
"""
SELECT "gender","modified","nickname","created","balance","is_admin","job_info","username","id" FROM "tableName" ORDER BY "is_admin" ASC
"""# return [dict] - values 可过滤需要的字段
await UserModel.all().values('id', 'username')
"""
SELECT "id" "id","username" "username" FROM "tableName" ORDER BY "is_admin" ASC
"""# 排序
await UserModel.all().order_by('-created')
"""
SELECT "job_info","id","gender","username","modified","nickname","created","balance","is_admin" FROM "tableName" ORDER BY "created" DESC
"""# 分页
await UserModel.all().offset(3).limit(2)
"""
SELECT "nickname","username","created","job_info","is_admin","balance","gender","modified","id" FROM "tableName" ORDER BY "is_admin" ASC LIMIT 2, 3
"""# 分组 value_list 可过滤需要字段 return ((字段值， )) 使用聚合函数 + 分组
await UserModel.all().annotate(result=Sum('balance')).group_by('gender').value_list('gender', 'result')
"""
SELECT "gender" "gender",SUM(CAST("balance" AS NUMERIC)) "result" FROM "tableName" GROUP BY "gender"
"""# 统计 返回 数字
await UserModel.all().count()
"""
SELECT COUNT(*) FROM "tableName"
"""

get

# 根据条件查询符合的对象，条件需唯一 return UserModel
await UserModel.get(nickname='777')
"""
SELECT "id","job_info","created","nickname","is_admin","balance","modified","gender","username" FROM "tableName" WHERE "username"='0' ORDER BY "is_admin" ASC LIMIT 2
"""# 查询不到返回None，避免 出现对象不存在的异常
await UserModel.get_or_none(username=100)
"""
SELECT "id","nickname","balance","created","modified","is_admin","username","gender","job_info" FROM "tableName" WHERE "username"='100' ORDER BY "is_admin" ASC LIMIT 2
"""# 如果有就返回查询数据，没有就创建
await UserModel.get_or_create(username=19, balance=22.9)
"""
SELECT "id","nickname","balance","created","modified","is_admin","username","gender","job_info" FROM "tableName" WHERE "username"='19' AND "balance"='22.9' ORDER BY "is_admin" ASC LIMIT 2
"""

filter 条件过滤

# return [UserModel] 条件查询 gender=1
await UserModel.filter(gender=1).count()
# 性别为1 的平均余额
await UserModel.filter(gender=1).annotate(price=Avg('balance')).values('price')
"""
SELECT AVG(CAST("balance" AS NUMERIC)) "price" FROM "tableName" WHERE "gender"=1
"""# 各性别平均余额
await UserModel.annotate(price=Avg('balance')).group_by('gender').values('gender', 'price')
"""
SELECT "gender" "gender",AVG(CAST("balance" AS NUMERIC)) "price" FROM "tableName" GROUP BY "gender"
"""# 获取第一个符合条件的
await UserModel.filter(nickname=777).first()
"""
SELECT "nickname","gender","is_admin","created","balance","job_info","id","username","modified" FROM "tableName" WHERE "nickname"='777' ORDER BY "is_admin" ASC LIMIT 1
"""# get sql
UserModel.filter(nickname=777).sql()
"""
SELECT "nickname","gender","is_admin","created","balance","job_info","id","username","modified" FROM "tableName" WHERE "nickname"='777' ORDER BY "is_admin" ASC"""# 查询 gender 不为 1
await UserModel.exclude(gender=1)
"""
SELECT "balance","modified","nickname","is_admin","id","created","job_info","username","gender" FROM "tableName" WHERE NOT "gender"=1 ORDER BY "is_admin" ASC"""# 查询 gender 不为 1
await UserModel.filter(gender__not=1)
"""
SELECT "nickname","job_info","is_admin","created","gender","balance","modified","username","id" FROM "tableName" WHERE "gender"<>1 OR "gender" IS NULL ORDER BY "is_admin" ASC"""# https://tortoise.github.io/query.html?h=__conta#filtering
# 包含 7 忽略大小写 -》 like
await UserModel.filter(nickname__icontains='7')
"""
SELECT "nickname","created","id","balance","is_admin","job_info","modified","gender","username" FROM "tableName" WHERE UPPER(CAST("nickname" AS VARCHAR)) LIKE UPPER('%7%') ESCAPE '\' ORDER BY "is_admin" ASC
"""# between 0 and 10
await UserModel.filter(balance__range=[0, 10])
"""
SELECT "gender","balance","is_admin","modified","nickname","id","job_info","created","username" FROM "tableName" WHERE "balance" BETWEEN 0 AND 10 ORDER BY "is_admin" ASC
"""# 大于等于1
await UserModel.filter(gender__gte=1)
"""
SELECT "username","modified","gender","job_info","id","balance","is_admin","nickname","created" FROM "tableName" WHERE "gender">=1 ORDER BY "is_admin" ASC
"""# is null
await UserModel.filter(gender__isnull=True)
"""
SELECT "username","modified","gender","job_info","id","balance","is_admin","nickname","created" FROM "tableName" WHERE "gender" IS NULL ORDER BY "is_admin" ASC
"""await UserModel.get_or_create(username=7809, balance=99.22, job_info={"breed": "labrador","owner": {"name": "Boby","last": None,"other_pets": [{"name": "Fishy",}],},})
# json 字段 owner 下other pets第一个 name 不为Fishy
await UserModel.filter(job_info__filter={"owner__other_pets__0__name__not": "Fishy"})
"""
SELECT "username","nickname","created","gender","job_info","balance","is_admin","id","modified" FROM "tableName" WHERE "username"='7809' AND "balance"='99.22' AND "job_info"='{"breed":"labrador","owner":{"name":"Boby","last":null,"other_pets":[{"name":"Fishy"}]}}' ORDER BY "is_admin" ASC LIMIT 2
"""# 子查询
await UserModel.filter(pk=Subquery(UserModel.filter(username=777).values("id"))).first()
"""
SELECT "job_info","created","username","is_admin","gender","balance","modified","nickname","id" FROM "tableName" WHERE "id"=(SELECT "id" "id" FROM "tableName" WHERE "username"='777' ORDER BY "is_admin" ASC) ORDER BY "is_admin" ASC LIMIT 1
"""

filter Q&F&RawSQL

# or
# username 为 '777' 或者 gender 不为1
await UserModel.filter(Q(username=777) | Q(gender__not=1))
"""
SELECT "created","id","job_info","username","gender","modified","balance","is_admin","nickname" FROM "tableName" WHERE "username"='777' OR "gender"<>1 OR "gender" IS NULL ORDER BY "is_admin" ASC
"""
await UserModel.filter(Q(username=777), Q(gender__not=1), join_type='OR')
"""
SELECT "created","id","job_info","username","gender","modified","balance","is_admin","nickname" FROM "tableName" WHERE "username"='777' AND ("gender"<>1 OR "gender" IS NULL) ORDER BY "is_admin" ASC
"""
# ~Q 否定
await UserModel.filter(Q(username=777), ~Q(gender=1), join_type='OR')
"""
SELECT "created","id","job_info","username","gender","modified","balance","is_admin","nickname" FROM "tableName" WHERE "username"='777' AND NOT "gender"=1 ORDER BY "is_admin" ASC
"""
# Q and 查询username 为 777 并且gender不为1
await UserModel.filter(Q(username=777), Q(gender__not=1))
"""
SELECT "username","gender","is_admin","job_info","balance","id","created","nickname","modified" FROM "tableName" WHERE "username"='777' AND ("gender"<>1 OR "gender" IS NULL) ORDER BY "is_admin" ASC
"""await UserModel.filter(Q(username=777), Q(gender__not=1), join_type='AND')
"""
SELECT "username","gender","is_admin","job_info","balance","id","created","nickname","modified" FROM "tableName" WHERE "username"='777' AND ("gender"<>1 OR "gender" IS NULL) ORDER BY "is_admin" ASC
"""await UserModel.filter(username=777, gender__not=1)
"""
SELECT "username","gender","is_admin","job_info","balance","id","created","nickname","modified" FROM "tableName" WHERE "username"='777' AND ("gender"<>1 OR "gender" IS NULL) ORDER BY "is_admin" ASC
"""# 将结果的 余额 + 999
await UserModel.annotate(idp=F("balance")+999).values("balance", "idp")
"""
SELECT "balance" "balance","balance"+999 "idp" FROM "tableName"
"""# RawSQL 余额 + 999, 可在其中执行SQL语句
await UserModel.annotate(idp=RawSQL("balance + 999")).values("balance", "idp")
"""
SELECT "balance" "balance",balance + 999 "idp" FROM "tableName"
"""# RawSQL 统计
await UserModel.annotate(idp=RawSQL("count(id)")).values('idp')
"""
SELECT count(id) "idp" FROM "tableName"
"""

原生SQL


# 执行原生SQL，返回字典
conn = connections.get("default")
await conn.execute_query_dict("SELECT COUNT(*) FROM tableName WHERE username=(?)", ['777'])
"""
SELECT COUNT(*) FROM tableName WHERE username='777'
"""

3、一对一关系

PS：1个学生仅有一个详细信息

模型定义

class Student(models.Model):name = fields.CharField(max_length=20, description="name")# 与下面的related_name 字段名一致，仅得到友好提示info: fields.ReverseRelation["StudentDetail"]class StudentDetail(models.Model):age = fields.IntField(description='age')# Student Object 通过 info拿到 StudentDetail Object, 类型标注友好提示 code联想补充students: fields.OneToOneRelation[Student] = fields.OneToOneField('models.Student', related_name='info')

表信息

CREATE TABLE "studentdetail" ("id" INTEGER PRIMARY KEY AUTOINCREMENT NOT NULL,"age" INT NOT NULL  /* age */,"student_id" INT NOT NULL UNIQUE REFERENCES "student" ("id") ON DELETE CASCADE
);
CREATE TABLE "student" ("id" INTEGER PRIMARY KEY AUTOINCREMENT NOT NULL,"name" VARCHAR(20) NOT NULL  /* name */
);

OneToOneField

model_name：相关模型的名称，格式为: samp: “{ app } . { model }”‘ format。
related_name：相关模型上的属性名，以反向解析外键
to_field: 对相关模型的属性名建立外键关系。如果未设置，则使用 pk, 关联模型上的某个建
on_delete：
- field.CASCADE:指示如果相关模型被删除，则该模型应级联删除。
- field.SET_DEFAULT:将字段重置为“ default”值，以防相关模型被删除。只能设置为字段具有“默认”设置。
- field.SET_NULL:如果相关模型被删除，则将字段重置为 NULL。只能在字段设置为“ null = True”时设置。
- field.RESTRICT:指示只要有外键指向相关模型删除，就会限制该删除。
db_constraint：默认值为 True，这几乎肯定是您想要的; 将其设置为 False 可能对数据完整性非常不利, 则为逻辑外建

CRUD

# create
student = await Student.create(name='7y')
await StudentDetail.create(age=18, students=student)
"""
INSERT INTO "student" ("name") VALUES ('7y')
# 15 为 student 的主键
INSERT INTO "studentdetail" ("age","students_id") VALUES (18,15)
"""# 改 只能改 某一张表的数据
await Student.filter(info__age=18).update(name='777')
"""
UPDATE "student" SET "name"=? FROM "student" "student_" LEFT OUTER JOIN "studentdetail" ON "student"."id"="studentdetail"."students_id" WHERE "studentdetail"."age"=18: ['777']
"""
await Student.filter(name='777').update(name='7y')
await StudentDetail.filter(students__name='777').update(age=18)# 查 学生名 7y 并且 学生 年龄 = 18 的数据 主表查从《关联表》表
res = await Student.filter(name='7y', info__age=18).first()
"""
SELECT "studentdetail"."age","studentdetail"."students_id","studentdetail"."id" FROM "studentdetail" LEFT OUTER JOIN "student" "studentdetail__students" ON "studentdetail__students"."id"="studentdetail"."students_id" WHERE "studentdetail__students"."name"='7y' LIMIT 1"""
# 通过res.related_name可获得被关联模型 StudentDetail
student_detail = await res.info.first()
print(student_detail.age) # 18res = (await StudentDetail.all().prefetch_related(Prefetch("students", queryset=Student.filter(name='7y'))).first())
print(await res.students.all().values())  # [{'id': 68, 'name': '7y'},]
"""
SELECT "students_id","age","id" FROM "studentdetail" LIMIT 1
SELECT "name","id" FROM "student" WHERE "name"='7y' AND "id" IN (68)
"""# 关联表查主表
res = await StudentDetail.filter(students__name='7y').first()
"""
SELECT "studentdetail"."age","studentdetail"."students_id","studentdetail"."id" FROM "studentdetail" LEFT OUTER JOIN "student" "studentdetail__students" ON "studentdetail__students"."id"="studentdetail"."students_id" WHERE "studentdetail__students"."name"='7y' LIMIT 1
"""
# 通过 res.students 获得 Student模型
student = await res.students.first()
print(student.name) # 7yres = await Student.filter(info__age=18).first().prefetch_related("info")
"""
SELECT "student"."name","student"."id" FROM "student" LEFT OUTER JOIN "studentdetail" ON "student"."id"="studentdetail"."students_id" WHERE "studentdetail"."age"=18 LIMIT 1
"""
print(res.info.age) # 18
"""
SELECT "students_id","age","id" FROM "studentdetail" WHERE "students_id" IN (68)
"""await StudentDetail.all().values('students__name', 'students__id', 'age')
"""
SELECT "studentdetail__students"."name" "students__name","studentdetail__students"."id" "students__id","studentdetail"."age" "age" FROM "studentdetail" LEFT OUTER JOIN "student" "studentdetail__students" ON "studentdetail__students"."id"="studentdetail"."students_id": None"""# 删 取决于 on_delete 策略
await Student.filter(name='7y').delete()

4、一对多关系

PS：1个班有多名学生；班级（1） -》学生（多），CRUD和 3. 一对一关系类似

模型定义


class Class(models.Model):id = fields.IntField(pk=True, description="班级唯一标识")name = fields.CharField(max_length=10, description="班级名称")# 类型提示，仅用于Code 联想字段 匹配 下方关联的related_namestudents: fields.ReverseRelation["Student"]class Student(models.Model):"""学生"""id = fields.IntField(pk=True, description="学生唯一标识")name = fields.CharField(max_length=20, description="学生名称")# Class Object 通过 students 拿到 List[Student Object] , 类型标注友好提示 code联想补充# 生成外建； 字段名 + _to_field（默认主表的主键）my_class: fields.ForeignKeyRelation[Class] = fields.ForeignKeyField("models.Class", related_name='students',description="所属班级")

表信息

CREATE TABLE "class" ("id" INTEGER PRIMARY KEY AUTOINCREMENT NOT NULL /* 班级唯一标识 */,"name" VARCHAR(10) NOT NULL  /* 班级名称 */
);CREATE TABLE "student" ("id" INTEGER PRIMARY KEY AUTOINCREMENT NOT NULL /* 学生唯一标识 */,"name" VARCHAR(20) NOT NULL  /* 学生名称 */,"my_class_id" INT NOT NULL REFERENCES "class" ("id") ON DELETE CASCADE /* 所属班级 */
);

ForeignKeyField

model_name：相关模型的名称，格式为: samp: “{ app } . { model }”‘ format。
related_name：相关模型上的属性名，以反向解析外键
to_field: 对相关模型的属性名建立外键关系。如果未设置，则使用 pk, 关联模型上的某个建
on_delete：
- - field.CASCADE:指示如果相关模型被删除，则该模型应级联删除。
  - field.RESTRICT:指示只要有外键指向相关模型删除，就会限制该删除。
  - field.SET_NULL:如果相关模型被删除，则将字段重置为 NULL。只能在字段设置为“ null = True”时设置。
  - field.SET_DEFAULT:将字段重置为“ default”值，以防相关模型被删除。只能设置为字段具有“默认”设置。
db_constraint：默认值为 True，这几乎肯定是您想要的; 将其设置为 False 可能对数据完整性非常不利, 则为逻辑外建

# create
class_obj = await Class.create(name='🚀班')
student_obj = await Student.create(name='7777777', my_class=class_obj)
# await Student.create(name='7777777', my_class=class_obj.id)
# await Student.create(name='7777777', my_class=class_obj.pk)
"""
INSERT INTO "class" ("name") VALUES (?): ['🚀班']
INSERT INTO "student" ("name","my_class_id") VALUES (?,?): ['7777777', 1]
"""# 通过表class（1） 获取 student（多）
for student in await class_obj.students.all():"""SELECT "id","name","my_class_id" FROM "student" WHERE "my_class_id"=4"""# pk 和 id 都是表示主键的意思print(student.name, student.id, student.pk)res = await Class.filter(id=3).first().prefetch_related("students")
print(await res.students.all().values())res = await Class.all().prefetch_related(Prefetch("students", queryset=Student.filter(my_class__name__not='7y').all())
)
print(await res[0].students.all().values())
"""
SELECT "student"."name","student"."id","student"."my_class_id" FROM "student" LEFT OUTER JOIN "class" "student__my_class" ON "student__my_class"."id"="student"."my_class_id" WHERE ("student__my_class"."name"<>'7y' OR "student__my_class"."name" IS NULL) AND "student"."my_class_id" IN (1,2,3,4,5,6)"""# 通过表 student（多） 获取 class（1）
print(await student_obj.my_class.all().values())
"""
SELECT "id" "id","name" "name" FROM "class"
"""res = await Student.filter(my_class__id__lt=2).all()
for cls in await res[0].my_class.all():print(cls)

5、多对多关系

模型定义


class Course(models.Model):"""课程表"""name = fields.CharField(max_length=20, description="课程名")students: fields.ManyToManyRelation["Student"]class Student(models.Model):name = fields.CharField(max_length=20)courses: fields.ManyToManyRelation[Course] = fields.ManyToManyField("models.Course",related_name="students")

表信息

CREATE TABLE "course" ("id" INTEGER PRIMARY KEY AUTOINCREMENT NOT NULL,"name" VARCHAR(20) NOT NULL  /* 课程名 */
);CREATE TABLE "student" ("id" INTEGER PRIMARY KEY AUTOINCREMENT NOT NULL,"name" VARCHAR(20) NOT NULL
);CREATE TABLE "student_course" ("student_id" INT NOT NULL REFERENCES "student" ("id") ON DELETE CASCADE,"course_id" INT NOT NULL REFERENCES "course" ("id") ON DELETE CASCADE
);

ManyToManyField

model_name：相关模型的名称，格式为: samp: “{ app } . { model }”‘ format。
related_name：相关模型上的属性名，以反向解析外键
to_field: 对相关模型的属性名建立外键关系。如果未设置，则使用 pk, 关联模型上的某个建
through：关联表的名称，默认当前表_关联表
forward_key：通过表的正向查找键。默认值通常是安全的。
backward_key：通过表的向后查找键。默认值通常是安全的。
on_delete：
- field.CASCADE:指示如果相关模型被删除，则该模型应级联删除。
  - field.RESTRICT:指示只要有外键指向相关模型删除，就会限制该删除。
  - field.SET_NULL:如果相关模型被删除，则将字段重置为 NULL。只能在字段设置为“ null = True”时设置。
  - field.SET_DEFAULT:将字段重置为“ default”值，以防相关模型被删除。只能设置为字段具有“默认”设置。
db_constraint：默认值为 True，这几乎肯定是您想要的; 将其设置为 False 可能对数据完整性非常不利, 则为逻辑外建

CRUD

# Create student object
student = await Student.create(name='7y')
"""
INSERT INTO "student" ("name") VALUES (?): ['7y']
"""
# 2. course object
course = await Course.create(name='Python')
"""
INSERT INTO "course" ("name") VALUES (?): ['Python']
"""
# 3. student_courses object
await student.courses.add(course)
"""
INSERT INTO "student_course" ("course_id","student_id") VALUES (3,3)
"""
await student.courses
"""
SELECT "course"."id","course"."name" FROM "course" LEFT OUTER JOIN "student_course" ON "course"."id"="student_course"."course_id" WHERE "student_course"."student_id"=5
"""# student表 获取course
for course in await student.courses:print(course.id) # 5r = await Student.filter(courses=5).prefetch_related("courses")
"""
SELECT "sq0"."_backward_relation_key" "_backward_relation_key","course"."name" "name","course"."id" "id" FROM "course" JOIN (SELECT "student_id" "_backward_relation_key","course_id" "_forward_relation_key" FROM "student_course" WHERE "student_id" IN (5)) "sq0" ON "sq0"."_forward_relation_key"="course"."id""""
print(r)# course表获取 student
await course.students
"""
SELECT "student"."name","student"."id" FROM "student" LEFT OUTER JOIN "student_course" ON "student"."id"="student_course"."student_id" WHERE "student_course"."course_id"=7
"""await student.fetch_related("courses")
await course.fetch_related("students")# 去掉重复
await Student.filter(courses__name='Python').distinct()
"""
SELECT DISTINCT "student"."id","student"."name" FROM "student" LEFT OUTER JOIN "student_course" ON "student"."id"="student_course"."student_id" LEFT OUTER JOIN "course" ON "student_course"."course_id"="course"."id" WHERE "course"."name"='Python'
"""

6、事务 & 信号 & 迁移

信号

ps：在CRUD前可添加处理逻辑

Simple Examples - Tortoise ORM v0.1

模型定义&信号注册

class User(models.Model):name = fields.CharField(max_length=30)age = fields.IntField()@pre_save(User)
async def signal_pre_save(sender: "Type[User]", instance: User, using_db, update_fields
) -> None:"""save 前执行"""print("pre save...", sender, instance, using_db, update_fields)@post_save(User)
async def signal_post_save(sender: "Type[User]",instance: User,created: bool,using_db: "Optional[BaseDBAsyncClient]",update_fields: List[str],
) -> None:"""save 后执行"""print("post.save...", sender, instance, using_db, created, update_fields)@pre_delete(User)
async def signal_pre_delete(sender: "Type[User]", instance: User, using_db: "Optional[BaseDBAsyncClient]"
) -> None:print("pre delete ...", sender, instance, using_db)@post_delete(User)
async def signal_post_delete(sender: "Type[User]", instance: User, using_db: "Optional[BaseDBAsyncClient]"
) -> None:print("post delete.", sender, instance, using_db)

操作


user = await User.create(name="23", age=18)
print("1\n")
user.name = "7y"
print("2\n")
await user.save(update_fields=["name"])
print("3\n")
await user.delete()
print("4\n")await User.create(name="231", age=19)# 下面 不触发
await User.filter(age=19).update(name="78")
await User.filter(age=19).delete()

事务

ps：事务中的sql某一个出现异常及回滚

Simple Examples - Tortoise ORM v0.22.0 Documentation

# 前置条件表中不包含age 30 的数据
# 当我们执行下面语句， 第一行将创建好数据， 第二行因为数据不存在 而rasie一个异常
await User.create(name="Test", age=19)
await User.get(age=30)  # 数据不存在 rasie Object does not exist# 这时候我们通过，这行更加笃定
await User.filter(name="Test", age=19).first()# 如果把他们放到一个事务中 则创建成功后回回滚，最终查不出数据 / 最终没写入到数据库

in_transaction 上下文管理器

# 1.事务方式一 in_transaction 上下文管理器， 可以debug查看更直观
try:async with in_transaction() as connection:# 1. 创建await User.create(name="Test", age=19)# 2. 创建是否成功print(await User.filter(name="Test", age=19).first())# 3. 报个错await User.get(age=30)  # 数据不存在 rasie Object does not exist
except OperationalError as e:print(e)
# 4. 确认是否回滚
print(await User.all())

atomic 装饰器

@atomic()
async def bound_to_fall():# 1. 创建await User.create(name="Test", age=19)# 2. 创建是否成功print(await User.filter(name="Test", age=19).first())# 3. 报个错
try:await bound_to_fall()
except OperationalError as e:print(e)
# 4. 确认是否回滚, 如果查到了 说明回滚失败 还是新增了数据
print(await User.filter(name="Test", age=19).first())

迁移

ps: 通过orm 生成 sql 再到表《按版本管理》

Migration - Tortoise ORM v0.22.0 Documentation

pip install aerich aiomysql# aerich -h 查看帮助文档downgrade  降级至指定版本heads      Show current available heads in migrate location.history    List all migrate items.init       Init config file and generate root migrate location.init-db    Generate schema and generate app migrate location.生成模式并生成应用程序迁移位置。migrate    Generate migrate changes file.生成迁移更改文件。upgrade    Upgrade to latest version.升级到最新版本。

TORTOISE_ORM = {# 连接信息"connections": {"default": "mysql://root:123456@127.0.0.1:3306/test"},"apps": {"models": {# model 信息# "models": ["more_more", "aerich.models", "table", "one_more"],"models": ["models", "aerich.models"], # 把需要的模型导进一个module 直接使用module"default_connection": "default",},},
}

# 1. 先初始化
aerich init -t conf.TORTOISE_ORM
# Success create migrate location ./migrations
# Success write config to pyproject.toml# 2. 初次使用生成表 和 迁移文件
aerich init-db
# Success create app migrate location migrations/models
# Success generate schema for app "models"# 3. 模型字段有变更时， 生成迁移文件
aerich migrate --name add_name_desc
# Success migrate 1_20221114234032_add_name_desc.py# 4. 迁移文件生成表
aerich upgrade
# Success upgrade 1_20221114234032_add_name_desc.py# 5. 迁移历史
aerich history
# 0_20221114233906_init.py
# 1_20221114234032_add_name_desc.py# 6. 回到上一次迁移
aerich downgrade
# Downgrade is dangerous, which maybe lose your data, are you sure? [y/N]:

7、主从数据库、读写分离

Router - Tortoise ORM v0.22.0 Documentation

1. 搭建MySQL主从
Docker部署Mysql8.0.20并配置主从复制 - C3Stones - 博客园
ps: m1芯片可以参考下面拉取镜像

m1 master

docker run --name mastermysql -d -p 3307:3306 -e MYSQL_ROOT_PASSWORD=123456 -v /Users/qy_air/mysql/master/data:/var/lib/mysql -v /Users/qy_air/mysql/master/conf/my.cnf:/etc/mysql/my.cnf mysql/mysql-server

m1 salve

docker run --name slavemysql -d -p 3308:3306 -e MYSQL_ROOT_PASSWORD=123456 -v /Users/qy_air/mysql/slave/data:/var/lib/mysql -v /Users/qy_air/mysql/slave/conf/my.cnf:/etc/mysql/my.cnf mysql/mysql-server

Tortoise 读写分离

"""
This example to use router to implement read/write separation
"""
from typing import Typefrom tortoise import Tortoise, fields, run_async
from tortoise.models import Modelclass Event(Model):id = fields.IntField(pk=True)name = fields.TextField()datetime = fields.DatetimeField(null=True)class Meta:table = "event"def __str__(self):return self.nameclass Router:def db_for_read(self, model: Type[Model]):return "slave"def db_for_write(self, model: Type[Model]):return "master"async def run():config = {"connections": {"master": "mysql://root:123456@127.0.0.1:3306/test","slave": "mysql://root:123456@127.0.0.1:3307/test"},"apps": {"models": {"models": ["__main__"],"default_connection": "master",}},"routers": ["__main__.Router"],"use_tz": False,"timezone": "UTC",}await Tortoise.init(config=config)await Tortoise.generate_schemas()# this will use connection masterevent = await Event.create(name="Test")# this will use connection slaveawait Event.get(pk=event.pk)if __name__ == "__main__":run_async(run())

8、序列化

因为这里的Pydantic我通常在FastAPI中有使用场景所以这里结合了pydantic 方便看序列化出来的模型效果

code

"""
This example demonstrates pydantic serialisation
已知问题：表中的关联字段 序列化 pydantic模型会被忽略，出现这种问题，可继承生成出来的schema 再把关系字段 写进去
"""
from typing import Optional, Listfrom tortoise import Tortoise, fields
from tortoise.contrib.pydantic import pydantic_model_creator, pydantic_queryset_creator
from tortoise.models import Modelasync def connect():await Tortoise.init(db_url="sqlite://:memory:", modules={"models": ["__main__"]})await Tortoise.generate_schemas()class User(Model):name = fields.CharField(max_length=20)loves = fields.JSONField(default=list)age = fields.IntField()password = fields.CharField(max_length=120)parent = fields.ForeignKeyField("models.User", related_name="pid", null=True, default=True)# 关系表序列化def length(self) -> int:"""计算属性 pydantic"""return len(self.name)# https://tortoise.github.io/contrib/pydantic.html?h=pydanticmeta#inheritanceclass PydanticMeta:allow_cycles: bool = False # 关系递归引用backward_relations: bool = True # 递归关系config_class = None # 自定义配置类exclude_raw_fields: bool = True # 排除原始的关系字段 字段包含_idmax_recursion: int = 3 # 最大递归级别sort_alphabetically: bool = False # 按字母排序# 生成模型排除字段exclude = ("created_at",)# Let's include two callables as computed columns 计算属性列computed = () # 计算属性，计算属性 必须带返回值类型Tortoise.init_models(["__main__"], "models")# model（ORM） to  schema（Pydantic） 序列化   ｜｜｜ schema（Pydantic） to  model（ORM） 反序列化# exclude 排除 include 包含Default = pydantic_model_creator(User, name="default")  # 所有字段
UserSchema = pydantic_model_creator(User, name="UserInput", exclude_readonly=True) # 只读 主键、
IncludeName = pydantic_model_creator(User, name="UserName", include=("name", ))
Computed = pydantic_model_creator(User, name="computed", computed=("length",))
UserSchemaOut = pydantic_model_creator(User, exclude=("password",))
UserSchemaList = pydantic_queryset_creator(User)from fastapi import FastAPI, APIRouterapp = FastAPI(title="Tortoise ORM Pydantic 序列化", on_startup=[connect], on_shutdown=[Tortoise.close_connections])serialization = APIRouter(tags=['model 序列化 pydantic - 需要model作为参数'])@serialization.post("/Default", summary="Default - 根据model全量")
def defa(data: Default):pass@serialization.post("/int", summary="UserSchema - 排除只读")
def inp(data: UserSchema):pass@serialization.post("/demo", summary="IncludeName - 仅包含name")
def ipn(data: IncludeName):pass@serialization.post("/Computed", summary="Computed - 添加计算属性")
def Comp(data: Computed):pass@serialization.post("/out", summary="UserSchemaOut - 排除password")
def out(data: UserSchemaOut):pass@serialization.post("/list", summary="UserSchemaList - 列表")
def li(data: UserSchemaList):passdeserialization = APIRouter(tags=["pydantic 反序列化 model - 需要model作为参数"])class Input(UserSchema):parent: Optional[int]@deserialization.post("/user", summary="from_tortoise_orm - 单个model")
async def add(data: Input):print(data)obj = await User.create(**data.dict())# 反序列化 需要一个Model 对象return await UserSchemaOut.from_tortoise_orm(obj)@deserialization.get("/user", summary="用户列表", response_model=UserSchemaList)
async def arr():# 反序列化 需要一个QuerySet 对象return await UserSchemaList.from_queryset(User.all())@deserialization.get("/user/{id}", summary="用户详细")
async def info(id: int):return await UserSchemaOut.from_queryset_single(User.filter(id=id).first())@deserialization.get("/user/q/{id}", summary="用户查询", response_model=List[UserSchemaOut])
async def query():return await UserSchemaOut.from_queryset(User.all())# return await UserSchemaList.from_queryset(User.all())# return await UserSchemaOut.from_queryset_single(User.all())app.include_router(serialization)
app.include_router(deserialization)if __name__ == '__main__':import uvicornuvicorn.run("__main__:app", reload=True)

序列化

@serialization.post("/Default", summary="Default - 根据model全量")
def defa(data: Default):pass"""
Default = pydantic_model_creator(User, name="default")  # 所有字段序列化的model
{"id": 2147483647,"name": "string","loves": "string","age": 2147483647,"password": "string"
}"""@serialization.post("/int", summary="UserSchema - 排除只读")
def inp(data: UserSchema):pass"""
UserSchema = pydantic_model_creator(User, name="UserInput", exclude_readonly=True)序列化的model{"name": "string","loves": "string","age": 2147483647,"password": "string"
}"""@serialization.post("/demo", summary="IncludeName - 仅包含name")
def ipn(data: IncludeName):pass"""
IncludeName = pydantic_model_creator(User, name="UserName", include=("name", )){"name": "string"
}"""@serialization.post("/Computed", summary="Computed - 添加计算属性")
def Comp(data: Computed):pass"""
Computed = pydantic_model_creator(User, name="computed", computed=("length",)){"id": 2147483647,"name": "string","loves": "string","age": 2147483647,"password": "string","length": 0
}
"""@serialization.post("/out", summary="UserSchemaOut - 排除password")
def out(data: UserSchemaOut):pass
"""
UserSchemaOut = pydantic_model_creator(User, exclude=("password",))
{"id": 2147483647,"name": "string","loves": "string","age": 2147483647
}"""@serialization.post("/list", summary="UserSchemaList - 列表")
def li(data: UserSchemaList):pass
"""
UserSchemaList = pydantic_queryset_creator(User)[{"id": 2147483647,"name": "string","loves": "string","age": 2147483647}
]
"""

models.Model 转 Pydantic Model

"""
参数
:param cls: The Tortoise Model
:param name: 显式指定自定义名称，而不是生成的名称<schmea 中展现>
:param exclude: 要从提供的模型中排除的额外字段。
:param include: 要从提供的模型中包括的额外字段。
:param computed: 从提供的模型中包含额外的计算字段（orm中不存在的）。
:param optional: 提供的模型的额外可选字段.
:param allow_cycles: Do we allow any cycles in the generated model?
我们在生成的模型中允许任何循环吗
这只对递归/自引用模型有用。值“ False”(默认值)将防止任何和所有的回溯。
This is only useful for recursive/self-referential models.A value of ``False`` (the default) will prevent any and all backtracking.
:param sort_alphabetically: Sort the parameters alphabetically instead of Field-definition order.
按字母顺序排序参数，而不是按字段定义顺序。
The default order would be:* Field definition order +
* order of reverse relations (as discovered) +
* order of computed functions (as provided).
:param exclude_readonly: Build a subset model that excludes any readonly fields
构建一个排除任何只读字段的子集模型
:param meta_override: A PydanticMeta class to override model's values.
用于覆盖模型值的 PydanticMeta 类。
"""UserSchema = pydantic_model_creator(User, exclude_readonly=True, exclude=("password",))print(UserSchema.schema())# 反序列化Modelobj = await User.create(name=123, age=18, passowrd=123)# 自动预取所有关系 如果不需要/或者想同步 则使用from_ormawait UserSchema.from_tortoise_orm(obj)UserSchema.from_orm(obj)

models.Model 转 List Pydantci Model

"""
参数
:param cls: The Tortoise Model
:param name: 显式指定自定义名称，而不是生成的名称<schmea 中展现>
:param exclude: 要从提供的模型中排除的额外字段。
:param include: 要从提供的模型中包括的额外字段。
:param computed: 从提供的模型中包含额外的计算字段（orm中不存在的）。
:param allow_cycles: Do we allow any cycles in the generated model?
我们在生成的模型中允许任何循环吗
这只对递归/自引用模型有用。值“ False”(默认值)将防止任何和所有的回溯。
This is only useful for recursive/self-referential models.A value of ``False`` (the default) will prevent any and all backtracking.
:param sort_alphabetically: Sort the parameters alphabetically instead of Field-definition order.
按字母顺序排序参数，而不是按字段定义顺序。
The default order would be:* Field definition order +
* order of reverse relations (as discovered) +
* order of computed functions (as provided).
"""UserSchemaList = pydantic_queryset_creator(User)print(UserSchemaList.schema())