Fastapi 与 Tortoise ORM，objs=UserM
Fastapi 与 Tortoise ORM，objs=UserM
1.Tortoise ORM 在项目中的配置
1.配置TORTOISE_ORM，参见：https://www.cnblogs.com/puffer/p/16428100.html
2.main.py主程序中进行注册
from tortoise.contrib.fastapi import register_tortoise
app = FastAPI()
@app.on_event("startup")
async def startup_event():
# generate_schemas=True 如果数据库为空，则自动生成对应表单，生产环境不要开
# add_exception_handlers=True 生产环境不要开，会泄露调试信息
register_tortoise(app, config=TORTOISE_ORM, generate_schemas=False, add_exception_handlers=False)
3.创建模型
from tortoise import Model
from tortoise import fields
class UserModel(DateTimeModel):
"""用户表"""
uuid = fields.UUIDField(source_field="uuid", pk=True, unique=True, description="用户唯一标识")
name = fields.CharField(source_field="name", max_length=30, description="用户名称")
age = fields.IntField(source_field="age", description="用户年龄")
password = fields.CharField(source_field="password", max_length=50, description="用户密码")
class Meta:
table = "user" # 表名
# abstract=True # 是否为抽象模型，用于继承，且不会在数据库中生成数据表
unique_together=(("name", "password"), ) # 联合约束
table_description = "用户表" # 数据库对该表的注释
indexes=(("name", "password"),) # 联合索引
ordering = ["xxx", "-xxx"] # 设置默认查询结果的顺序，-代表降序
2.Tortoise ORM 增删改查
1.查询
-
1.使用 await 和 不使用 await 的区别
- 1.使用
await 查询时，得到的结果是Model对象，可以直接获取其属性objs = await UserModel.filter(uuid__in=apps_uuid)
# {VirtualAppModel} <VirtualAppModel>
print(objs[0].name)
- 2.不适用
await 查询时，得到的是tortoise.queryset.QuerySet对象，无法通过循环遍历获取每一个对象的具体属性，但是支持链式查询，适合在进行分页时使用。objs = UserModel.filter(uuid__in=apps_uuid)
# obj: <tortoise.queryset.QuerySet object at 0x00000132C4EBF160>
objs = objs.filter(name=xxx)
objs = objs.filter(xxx=xxxx)
-
2.判断对象是否存在
result = await UserModel.exists(uuid=xxx)
-
3.获取单条数据
user = await UserModel.get(uuid=xxx) # <UserModel>
# 1.如果查询到多条数据，则抛出异常：tortoise.exceptions.MultipleObjectsReturned
# 2.如果没找到数据，则抛出异常：tortoise.exceptions.DoesNotExist
user = await UserModel.get_or_none(uuid=xxx) # <UserModel> or None
# 1.如果没找到数据，返回：None
# 2.如果查询到多条数据，则抛出异常：tortoise.exceptions.MultipleObjectsReturned
-
3.获取多条数据
users = await UserModel.filter(name=xxx) # [<UserModel>]
-
4.获取所有数据
users = await UserModel.all() # [<UserModel>, <UserModel>, ...]
-
5.获取第一条数据
data = await UserModel.first() # [<UserModel>]
-
6.仅获取模型中部分的字段
data_dict = await UserModel.first().values("name", "uuid")
# 如果查询结果是单条数据：{'name': '222224', 'uuid': '35f01c8a57aa44008c99682f0eece37a'}
# 如果查询结果是多条数据：[{'name': 'xxx', 'uuid': 'xxx'}, {'name': 'xxx', 'uuid': 'xxx'}]
data_tuple = await UserModel.first().values_list("name", "uuid")
# 元组形式，只返回值：('222224', '35f01c8a57aa44008c99682f0eece37a')
# 多条数据：[('222224', '35f01c8a57aa44008c99682f0eece37a'), ('xxx', 'xxx')]
# 如果相获取多条数据中的某一个字段，比如uuid，正常情况下返回：[('xxx',), ('xxx'),...]
# 但如果想要的结果为：['xxx', 'xxx', 'xxx', ...]，就需要另一个参数了
uuid_list = await UserModel.filter(name__startswith='xxx').values_list("uuid", flat=True)
# 如果想获取部分字段，但是以object对象形式返回
data = await VirtualAppModel.first().only("name")
# 如果查询结果是单条数据：<UserModel>，不过这个模型对象中只有name属性，强行获取其他属性，则会报错
# 如果查询结果是多条数据：[<UserModel>]
-
7.select_related、prefetch_related，常用于关联模型查询中，减少数据库访问频次，提高查询性能，此处不多做演示，具体可以查看django orm中的示例https://blog.csdn.net/qq_42517220/article/details/93381250
-
8.不等于：exclude，比如查询name不等于111的用户
data = await UserModel.exclude(name='111')
-
9.数据去重
data = await UserModel.filter(name='111').distinct()
-
10.统计条数
num = await UserModel.filter(name='test').count()
# 或者
queryset = UserModel.filter(name='test')
num = await queryset.count()
data = await queryset.all()
-
11.聚合查询
from tortoise.functions import Count, Trim, Lower, Upper, Coalesce
await Tournament.annotate(events_count=Count('events')).filter(events_count__gte=10)
await Tournament.annotate(clean_name=Trim('name')).filter(clean_name='tournament')
await Tournament.annotate(name_upper=Upper('name')).filter(name_upper='TOURNAMENT')
await Tournament.annotate(name_lower=Lower('name')).filter(name_lower='tournament')
await Tournament.annotate(desc_clean=Coalesce('desc', '')).filter(desc_clean='')
-
12.双下划线查询：根据字段值进行过滤
关键字
意义
使用方法
not
不等于
name__not='xxx'
in
范围内
name__in=['xxx', 'xxx']
not_in
范围外
name__not_in=['xxx', 'xxx']
gte
大于或等于
age__gte=22
gt
大于
age__gt=22
lte
小于等于
age__lte=45
lte
小于
age__lt=45
range
范围查询
age__range=(18,45)
isnull
null查询
desc__isnull=True
not_isnull
非null查询
desc__not_isnull=True
contains
包含查询
name__contains="test"
icontains
不区分大小写包含查询
name__icontains="test"
startswith
开头查询
name__startswith="test"
istartswith
不区分大小写开头查询
name__istartswith="test"
endswith
结尾查询
name__endswith="test"
iendswith
不区分大小写结尾查询
name__iendswith="test"
iexact
不区分大小写的等于
name__iexact="test"
search
全文检索（测试报错）
name__search="test"
year
年份查询
created_at__year=2020
month
月份查询
created_at__month=7
day
日查询
created_at__day=24
-
13.JSON类型数据查询
# 1.列表形式
# extra字段值：["text", 3, {"msg": "msg2"}]
obj = await UserModel.filter(extra__contains=[{"msg": "msg2"}]).first()
# 2.字典形式
# extra字段值：{"breed": "labrador", owner": {"name": "Boby", "last": None, other_pets": [{"name": "Fishy"}]}}
# 2.1根据字段进行精确匹配（可以使用上述双下划线查询，比如：name__not进行不等于查询）
obj1 = await UserModel.filter(extra__filter={"breed": "labrador"}).first()
# 2.2嵌套字典数据获取
obj2 = await UserModel.filter(extra__filter={"owner__name": "Boby"}).first()
# 2.3获取嵌套数据中的列表数据
obj3 = await UserModel.filter(data__filter={"owner__other_pets__0__name": "Fishy"}).first()
-
14.Q查询
from tortoise.expressions import Q
users = await UserModel.filter(Q(name='a') | Q(name='b'))
# 等效于：如果省略join_type，则为AND
users = await UserModel.filter(Q(Q(name='a'), Q(name='b'), join_type="OR"))
# ~Q：不等于
users = await UserModel.filter(~Q(name='a'))
-
15.F表达式
from tortoise.expressions import F
await UserModel.filter(uuid='xxx').update(age=F('age')-10)
2.创建
- 1.单条数据创建
user = await UserModel.create(uuid="xxx", name="xxx") # <UserModel>
# 或者
user = User(uuid="xxx", name="xxx")
await user.save()
- 2.批量创建
bulk_data = list()
for uuid in users_uuid:
bulk_data.append(await UserModel(uuid=uuid, name="xxx"))
await UserModel.bulk_create(bulk_data)
- 3.查询或创建（如果查询得到，则返回该对象，否则直接创建）
# 根据名称查询，查询不到，则创建一条数据：name:xxx，age: 12
user = await UserModel.get_or_create(name='xxx', defaults={'age': 12})
3.更新
- 1.根据条件更新
# 返回更新的数目：int
num = await UserModel.filter(uuid='xxx').update(age=24)
# 或者
user = await UserModel.get(uuid='xxx')
user.age = F('age') + 10
await user.save(update_fields=['balance'])
- 2.更新还是创建
# 根据name=111查询，如果有数据，则更新age=26，password=password；否则，创建一条数据，name='11111'、age=26、password=password
# 返回结果为元组，第一个元素为Model对象，第二个元素为是否新创建的数据
data, is_create = await UserModel.update_or_create(name="11111", defaults={"age": 26, "password": "password"})
4.删除
- 1.删除全部
# 删除的条目数
num = await UserModel.all().delete()
- 2.根据条件删除
# 删除的条目数
num = await UserModel.filter(uuid='xxx').delete()
from tortoise.contrib.fastapi import register_tortoise
app = FastAPI()
@app.on_event("startup")
async def startup_event():
# generate_schemas=True 如果数据库为空，则自动生成对应表单，生产环境不要开
# add_exception_handlers=True 生产环境不要开，会泄露调试信息
register_tortoise(app, config=TORTOISE_ORM, generate_schemas=False, add_exception_handlers=False)
from tortoise import Model
from tortoise import fields
class UserModel(DateTimeModel):
"""用户表"""
uuid = fields.UUIDField(source_field="uuid", pk=True, unique=True, description="用户唯一标识")
name = fields.CharField(source_field="name", max_length=30, description="用户名称")
age = fields.IntField(source_field="age", description="用户年龄")
password = fields.CharField(source_field="password", max_length=50, description="用户密码")
class Meta:
table = "user" # 表名
# abstract=True # 是否为抽象模型，用于继承，且不会在数据库中生成数据表
unique_together=(("name", "password"), ) # 联合约束
table_description = "用户表" # 数据库对该表的注释
indexes=(("name", "password"),) # 联合索引
ordering = ["xxx", "-xxx"] # 设置默认查询结果的顺序，-代表降序
1.查询
-
1.使用
await和 不使用await的区别- 1.使用
await查询时，得到的结果是Model对象，可以直接获取其属性objs = await UserModel.filter(uuid__in=apps_uuid) # {VirtualAppModel} <VirtualAppModel> print(objs[0].name) - 2.不适用
await查询时，得到的是tortoise.queryset.QuerySet对象，无法通过循环遍历获取每一个对象的具体属性，但是支持链式查询，适合在进行分页时使用。objs = UserModel.filter(uuid__in=apps_uuid) # obj: <tortoise.queryset.QuerySet object at 0x00000132C4EBF160> objs = objs.filter(name=xxx) objs = objs.filter(xxx=xxxx)
- 1.使用
-
2.判断对象是否存在
result = await UserModel.exists(uuid=xxx) -
3.获取单条数据
user = await UserModel.get(uuid=xxx) # <UserModel> # 1.如果查询到多条数据，则抛出异常：tortoise.exceptions.MultipleObjectsReturned # 2.如果没找到数据，则抛出异常：tortoise.exceptions.DoesNotExist user = await UserModel.get_or_none(uuid=xxx) # <UserModel> or None # 1.如果没找到数据，返回：None # 2.如果查询到多条数据，则抛出异常：tortoise.exceptions.MultipleObjectsReturned -
3.获取多条数据
users = await UserModel.filter(name=xxx) # [<UserModel>] -
4.获取所有数据
users = await UserModel.all() # [<UserModel>, <UserModel>, ...] -
5.获取第一条数据
data = await UserModel.first() # [<UserModel>] -
6.仅获取模型中部分的字段
data_dict = await UserModel.first().values("name", "uuid") # 如果查询结果是单条数据：{'name': '222224', 'uuid': '35f01c8a57aa44008c99682f0eece37a'} # 如果查询结果是多条数据：[{'name': 'xxx', 'uuid': 'xxx'}, {'name': 'xxx', 'uuid': 'xxx'}] data_tuple = await UserModel.first().values_list("name", "uuid") # 元组形式，只返回值：('222224', '35f01c8a57aa44008c99682f0eece37a') # 多条数据：[('222224', '35f01c8a57aa44008c99682f0eece37a'), ('xxx', 'xxx')] # 如果相获取多条数据中的某一个字段，比如uuid，正常情况下返回：[('xxx',), ('xxx'),...] # 但如果想要的结果为：['xxx', 'xxx', 'xxx', ...]，就需要另一个参数了 uuid_list = await UserModel.filter(name__startswith='xxx').values_list("uuid", flat=True) # 如果想获取部分字段，但是以object对象形式返回 data = await VirtualAppModel.first().only("name") # 如果查询结果是单条数据：<UserModel>，不过这个模型对象中只有name属性，强行获取其他属性，则会报错 # 如果查询结果是多条数据：[<UserModel>] -
7.
select_related、prefetch_related，常用于关联模型查询中，减少数据库访问频次，提高查询性能，此处不多做演示，具体可以查看django orm中的示例https://blog.csdn.net/qq_42517220/article/details/93381250 -
8.不等于：exclude，比如查询name不等于111的用户
data = await UserModel.exclude(name='111') -
9.数据去重
data = await UserModel.filter(name='111').distinct() -
10.统计条数
num = await UserModel.filter(name='test').count() # 或者 queryset = UserModel.filter(name='test') num = await queryset.count() data = await queryset.all() -
11.聚合查询
from tortoise.functions import Count, Trim, Lower, Upper, Coalesce await Tournament.annotate(events_count=Count('events')).filter(events_count__gte=10) await Tournament.annotate(clean_name=Trim('name')).filter(clean_name='tournament') await Tournament.annotate(name_upper=Upper('name')).filter(name_upper='TOURNAMENT') await Tournament.annotate(name_lower=Lower('name')).filter(name_lower='tournament') await Tournament.annotate(desc_clean=Coalesce('desc', '')).filter(desc_clean='') -
12.双下划线查询：根据字段值进行过滤
关键字 意义 使用方法 not 不等于 name__not='xxx' in 范围内 name__in=['xxx', 'xxx'] not_in 范围外 name__not_in=['xxx', 'xxx'] gte 大于或等于 age__gte=22 gt 大于 age__gt=22 lte 小于等于 age__lte=45 lte 小于 age__lt=45 range 范围查询 age__range=(18,45) isnull null查询 desc__isnull=True not_isnull 非null查询 desc__not_isnull=True contains 包含查询 name__contains="test" icontains 不区分大小写包含查询 name__icontains="test" startswith 开头查询 name__startswith="test" istartswith 不区分大小写开头查询 name__istartswith="test" endswith 结尾查询 name__endswith="test" iendswith 不区分大小写结尾查询 name__iendswith="test" iexact 不区分大小写的等于 name__iexact="test" search 全文检索（测试报错） name__search="test" year 年份查询 created_at__year=2020 month 月份查询 created_at__month=7 day 日查询 created_at__day=24 -
13.
JSON类型数据查询# 1.列表形式 # extra字段值：["text", 3, {"msg": "msg2"}] obj = await UserModel.filter(extra__contains=[{"msg": "msg2"}]).first() # 2.字典形式 # extra字段值：{"breed": "labrador", owner": {"name": "Boby", "last": None, other_pets": [{"name": "Fishy"}]}} # 2.1根据字段进行精确匹配（可以使用上述双下划线查询，比如：name__not进行不等于查询） obj1 = await UserModel.filter(extra__filter={"breed": "labrador"}).first() # 2.2嵌套字典数据获取 obj2 = await UserModel.filter(extra__filter={"owner__name": "Boby"}).first() # 2.3获取嵌套数据中的列表数据 obj3 = await UserModel.filter(data__filter={"owner__other_pets__0__name": "Fishy"}).first() -
14.
Q查询from tortoise.expressions import Q users = await UserModel.filter(Q(name='a') | Q(name='b')) # 等效于：如果省略join_type，则为AND users = await UserModel.filter(Q(Q(name='a'), Q(name='b'), join_type="OR")) # ~Q：不等于 users = await UserModel.filter(~Q(name='a')) -
15.
F表达式from tortoise.expressions import F await UserModel.filter(uuid='xxx').update(age=F('age')-10)
2.创建
- 1.单条数据创建
user = await UserModel.create(uuid="xxx", name="xxx") # <UserModel> # 或者 user = User(uuid="xxx", name="xxx") await user.save() - 2.批量创建
bulk_data = list() for uuid in users_uuid: bulk_data.append(await UserModel(uuid=uuid, name="xxx")) await UserModel.bulk_create(bulk_data) - 3.查询或创建（如果查询得到，则返回该对象，否则直接创建）
# 根据名称查询，查询不到，则创建一条数据：name:xxx，age: 12 user = await UserModel.get_or_create(name='xxx', defaults={'age': 12})
3.更新
- 1.根据条件更新
# 返回更新的数目：int num = await UserModel.filter(uuid='xxx').update(age=24) # 或者 user = await UserModel.get(uuid='xxx') user.age = F('age') + 10 await user.save(update_fields=['balance']) - 2.更新还是创建
# 根据name=111查询，如果有数据，则更新age=26，password=password；否则，创建一条数据，name='11111'、age=26、password=password # 返回结果为元组，第一个元素为Model对象，第二个元素为是否新创建的数据 data, is_create = await UserModel.update_or_create(name="11111", defaults={"age": 26, "password": "password"})
4.删除
- 1.删除全部
# 删除的条目数 num = await UserModel.all().delete() - 2.根据条件删除
# 删除的条目数 num = await UserModel.filter(uuid='xxx').delete()
评论关闭