说明
经过fastzdp_login的整合,我们发现,fastzdp_sqlmodel还可以继续封装两个便捷的方法。
- get_first:获取查询结果集中的第一条数据
- is_exitsts:判断数据是否已存在
封装get_first方法
def get_first(engine, model, query_dict: dict):
"""
获取查询结果中的第一条数据
:param engine: 引擎对象
:param model: 模型类
:param query_dict: 查询字典
:return: 存在数据返回模型对象,否则返回None
"""
data = get_by_dict(engine, model, query_dict)
if isinstance(data, list) and len(data) > 0:
return data[0]
测试get_first方法
示例:查询名字叫张三1的用户
from typing import Optional
from sqlmodel import Field, SQLModel
import fastzdp_sqlmodel as fs
class User(SQLModel, table=True):
id: Optional[int] = Field(default=None, primary_key=True)
name: str
age: Optional[int] = None
# 创建数据库引擎
engine = fs.get_engine(database="fastzdp_sqlmodel")
# 初始化表格
fs.init_table(engine)
# 创建数据
fs.add(engine, User(name="张三1", age=23))
fs.add(engine, User(name="张三2", age=23))
fs.add(engine, User(name="张三3", age=23))
# 查询第一条数据
u = fs.get_first(engine, User, {"name": "张三1"})
print(u)
执行结果如下:
从测试结果来看,这个方法是基本符合预期的。
封装is_exists方法
这个方法比较简单,我们只需要执行查询,有查询结果即可。
def is_exists(engine, model, query_dict: dict) -> bool:
"""
根据查询字典指定的键值对等值查询数据
:param engine: 引擎对象
:param model: 模型类
:param query_dict: 查询字典
"""
data = get_by_dict(engine, model, query_dict)
return isinstance(data, list) and len(data) > 0
测试is_exists方法
我们分别判断一个存在的数据和一个不存在的数据,查看结果是否符合预期。
from typing import Optional
from sqlmodel import Field, SQLModel
import fastzdp_sqlmodel as fs
class User(SQLModel, table=True):
id: Optional[int] = Field(default=None, primary_key=True)
name: str
age: Optional[int] = None
# 创建数据库引擎
engine = fs.get_engine(database="fastzdp_sqlmodel")
# 初始化表格
fs.init_table(engine)
# 创建数据
fs.add(engine, User(name="张三1", age=23))
fs.add(engine, User(name="张三2", age=23))
fs.add(engine, User(name="张三3", age=23))
# 判断是否存在
print(fs.is_exists(engine, User, {"name": "张三1"}))
print(fs.is_exists(engine, User, {"name": "张三11"}))
执行结果如下:
从执行结果来看,是基本符合预期的。
总结
本教程主要记录了get_first和is_exists这两个方法的封装过程。
如果大家需要FastAPI零基础到项目实战的全套录播课程,或者Python一对一的私教课指导,欢迎留言或者私信。