关于SQLModel库的一些封装思路和实战，fastzdp_sqlmodel库封装过程中的一些笔记，20240817

news2024/11/14 15:28:40

概述

最近在对SQLModel做一些封装，本篇文章主要记录封装过程中的一些思路和实战代码。

实现or查询

原本的代码

from typing import Optional
from sqlmodel import Field, SQLModel, create_engine, Session, select, or_


class Hero(SQLModel, table=True):
    id: Optional[int] = Field(default=None, primary_key=True)
    name: str
    secret_name: str
    age: Optional[int] = None


# 创建数据库引擎
sqlite_url = "mysql+pymysql://root:root@127.0.0.1:3306/fastzdp_sqlmodel?charset=utf8mb4"
engine = create_engine(sqlite_url, echo=True)

# 创建所有表
SQLModel.metadata.drop_all(engine)
SQLModel.metadata.create_all(engine)

# 新增
hero_1 = Hero(name="Deadpond", secret_name="Dive Wilson")
hero_2 = Hero(name="Spider-Boy", secret_name="Pedro Parqueador")
hero_3 = Hero(name="Rusty-Man", secret_name="Tommy Sharp", age=48)
hero_4 = Hero(name="Tarantula", secret_name="Natalia Roman-on", age=32)
hero_5 = Hero(name="Black Lion", secret_name="Trevor Challa", age=35)
hero_6 = Hero(name="Dr. Weird", secret_name="Steve Weird", age=36)
hero_7 = Hero(name="Captain North America", secret_name="Esteban Rogelios", age=93)
with Session(engine) as session:
    session.add(hero_1)
    session.add(hero_2)
    session.add(hero_3)
    session.add(hero_4)
    session.add(hero_5)
    session.add(hero_6)
    session.add(hero_7)
    session.commit()

# or 查询
with Session(engine) as session:
    statement = select(Hero).where(or_(Hero.age <= 35, Hero.age > 90))
    results = session.exec(statement)
    for hero in results:
        print(hero)

思考

这个代码用起来虽然已经比较简单了，但是我觉得还可以更简单，更灵活，所以想要进一步封装一个方法，能够轻松的支持or查询。

初步设计

利用反射的一些特性，我初步做了如下方法的一个封装。

def get_by_dict_or(engine, model, query_dict: dict):
    """
    根据查询字典指定的键值对查询数据，键值对之间的关系是or的关系
    :param engine: 引擎对象
    :param model: 模型类
    :param query_dict: 查询字典 {age:{gt:3,lt:33}}
    """
    with Session(engine) as session:
        query = select(model)
        if isinstance(query_dict, dict):
            for k, v in query_dict.items():
                if hasattr(model, k):
                    if isinstance(v, dict):
                        # or_(Hero.age <= 35, Hero.age > 90)
                        mk = getattr(model, k)
                        conditions = []
                        for kk, vv in v.items():
                            if kk == "==":
                                conditions.append(mk == vv)
                            elif kk == ">":
                                conditions.append(mk > vv)
                            elif kk == "<":
                                conditions.append(mk < vv)
                            elif kk == ">=":
                                conditions.append(mk >= vv)
                            elif kk == "<=":
                                conditions.append(mk <= vv)
                        query = query.where(or_(*conditions))
        results = session.exec(query)
        return results.all()

初步测试

这个方法能不能行呢？

先写个简单的测试代码测一下。

from typing import Optional
from sqlmodel import Field, SQLModel
import fastzdp_sqlmodel as fsqlmodel


class User(SQLModel, table=True):
    id: Optional[int] = Field(default=None, primary_key=True)
    name: str
    age: Optional[int] = None


# 创建数据库引擎
engine = fsqlmodel.get_engine(database="fastzdp_sqlmodel")

# 初始化表格
fsqlmodel.init_table(engine)

# 创建数据
u = User(name="张三", age=23)
fsqlmodel.add(engine, u)
u = User(name="张三", age=33)
fsqlmodel.add(engine, u)
u = User(name="李四", age=24)
fsqlmodel.add(engine, u)

# 查询年龄等于23或者小于10的用户
query = {"age": {">": 23, "<": 10}}
users = fsqlmodel.get_by_dict_or(engine, User, query)
print(users)

执行结果如下：
在这里插入图片描述

初步来看是比较符合预期的。

突然的想法

这种or字典查询的方式看上去比较易懂，能不能支持and查询呢。

比如 {age:{>:20, <:30}} 不就是一个非常容易看懂的range查询吗？

所以我决定对上面的这个方法做一个改造，封装另一个方法。

初步方法封装

将之前的方法复制过来改个名就行了。

def get_by_dict_and(engine, model, query_dict: dict):
    """
    根据查询字典指定的键值对查询数据，键值对之间的关系是and的关系
    :param engine: 引擎对象
    :param model: 模型类
    :param query_dict: 查询字典 {age:{gt:3,lt:33}}
    """
    with Session(engine) as session:
        query = select(model)
        if isinstance(query_dict, dict):
            for k, v in query_dict.items():
                if hasattr(model, k):
                    if isinstance(v, dict):
                        # or_(Hero.age <= 35, Hero.age > 90)
                        mk = getattr(model, k)
                        conditions = []
                        for kk, vv in v.items():
                            if kk == "==":
                                conditions.append(mk == vv)
                            elif kk == ">":
                                conditions.append(mk > vv)
                            elif kk == "<":
                                conditions.append(mk < vv)
                            elif kk == ">=":
                                conditions.append(mk >= vv)
                            elif kk == "<=":
                                conditions.append(mk <= vv)
                        query = query.where(*conditions)
        results = session.exec(query)
        return results.all()

稍微测试一下

也是在上一个测试代码的基础上稍微改造一下。

from typing import Optional
from sqlmodel import Field, SQLModel
import fastzdp_sqlmodel as fsqlmodel


class User(SQLModel, table=True):
    id: Optional[int] = Field(default=None, primary_key=True)
    name: str
    age: Optional[int] = None


# 创建数据库引擎
engine = fsqlmodel.get_engine(database="fastzdp_sqlmodel")

# 初始化表格
fsqlmodel.init_table(engine)

# 创建数据
u = User(name="张三", age=23)
fsqlmodel.add(engine, u)
u = User(name="张三", age=33)
fsqlmodel.add(engine, u)
u = User(name="李四", age=24)
fsqlmodel.add(engine, u)

# 查询年龄等于20或者小于30的用户
query = {"age": {">": 20, "<": 30}}
users = fsqlmodel.get_by_dict_and(engine, User, query)
print(users)

输出结果如下：
在这里插入图片描述

初步来看，也是非常符合预期的。

遗留问题

实际上上面的代码还有不少遗留的问题，比如我想要查询名字是张三或者年龄大于20的怎么查询呢？

不过我们有了初步的方法以后，后面就是增加功能和优化了，暂时不急。

我决定先把整个框架的大体架子给搭建起来。

封装根据ID查询的方法

原本的代码1

from typing import Optional
from sqlmodel import Field, SQLModel, create_engine, Session, select, or_


class Hero(SQLModel, table=True):
    id: int | None = Field(default=None, primary_key=True)
    name: str = Field(index=True)
    secret_name: str
    age: int | None = Field(default=None, index=True)


# 创建数据库引擎
sqlite_url = "mysql+pymysql://root:root@127.0.0.1:3306/fastzdp_sqlmodel?charset=utf8mb4"
engine = create_engine(sqlite_url, echo=True)

# 创建所有表
SQLModel.metadata.drop_all(engine)
SQLModel.metadata.create_all(engine)

# 新增
hero_1 = Hero(name="Deadpond", secret_name="Dive Wilson")
hero_2 = Hero(name="Spider-Boy", secret_name="Pedro Parqueador")
hero_3 = Hero(name="Rusty-Man", secret_name="Tommy Sharp", age=48)
hero_4 = Hero(name="Tarantula", secret_name="Natalia Roman-on", age=32)
hero_5 = Hero(name="Black Lion", secret_name="Trevor Challa", age=35)
hero_6 = Hero(name="Dr. Weird", secret_name="Steve Weird", age=36)
hero_7 = Hero(name="Captain North America", secret_name="Esteban Rogelios", age=93)
with Session(engine) as session:
    session.add(hero_1)
    session.add(hero_2)
    session.add(hero_3)
    session.add(hero_4)
    session.add(hero_5)
    session.add(hero_6)
    session.add(hero_7)
    session.commit()

# 查询
with Session(engine) as session:
    statement = select(Hero).where(Hero.id == 1)
    results = session.exec(statement)
    hero = results.first()
    print("Hero:", hero)

原本的代码2

from typing import Optional
from sqlmodel import Field, SQLModel, create_engine, Session, select, or_


class Hero(SQLModel, table=True):
    id: int | None = Field(default=None, primary_key=True)
    name: str = Field(index=True)
    secret_name: str
    age: int | None = Field(default=None, index=True)


# 创建数据库引擎
sqlite_url = "mysql+pymysql://root:root@127.0.0.1:3306/fastzdp_sqlmodel?charset=utf8mb4"
engine = create_engine(sqlite_url, echo=True)

# 创建所有表
SQLModel.metadata.drop_all(engine)
SQLModel.metadata.create_all(engine)

# 新增
hero_1 = Hero(name="Deadpond", secret_name="Dive Wilson")
hero_2 = Hero(name="Spider-Boy", secret_name="Pedro Parqueador")
hero_3 = Hero(name="Rusty-Man", secret_name="Tommy Sharp", age=48)
hero_4 = Hero(name="Tarantula", secret_name="Natalia Roman-on", age=32)
hero_5 = Hero(name="Black Lion", secret_name="Trevor Challa", age=35)
hero_6 = Hero(name="Dr. Weird", secret_name="Steve Weird", age=36)
hero_7 = Hero(name="Captain North America", secret_name="Esteban Rogelios", age=93)
with Session(engine) as session:
    session.add(hero_1)
    session.add(hero_2)
    session.add(hero_3)
    session.add(hero_4)
    session.add(hero_5)
    session.add(hero_6)
    session.add(hero_7)
    session.commit()

# 查询
with Session(engine) as session:
    hero = session.get(Hero, 1)
    print("Hero:", hero)

简单的思考

如果ID不存在，需不需要抛出异常？不需要

所以我这里决定选first这个方法。

初步封装

def get(engine, model, id):
    """
    根据ID查询数据
    :param engine: 连接数据库的引擎对象
    :param model: 模型类
    :param id: 要查找的id的值
    :return: id对应的模型对象，如果不存在返回None
    """
    with Session(engine) as session:
        return session.exec(select(model).where(model.id == id)).first()

简单的测试

测试代码如下：

from typing import Optional
from sqlmodel import Field, SQLModel
import fastzdp_sqlmodel as fsqlmodel


class User(SQLModel, table=True):
    id: Optional[int] = Field(default=None, primary_key=True)
    name: str
    age: Optional[int] = None


# 创建数据库引擎
engine = fsqlmodel.get_engine(database="fastzdp_sqlmodel")

# 初始化表格
fsqlmodel.init_table(engine)

# 创建数据
u = User(name="张三", age=23)
fsqlmodel.add(engine, u)
u = User(name="张三", age=33)
fsqlmodel.add(engine, u)
u = User(name="李四", age=24)
fsqlmodel.add(engine, u)

# 查询id为1的数据
u = fsqlmodel.get(engine, User, 1)
print(u)

# 查询id为11的数据
u = fsqlmodel.get(engine, User, 11)
print(u)

输出结果：
在这里插入图片描述

从输出结果来看，这个方法是比较符合预期的。

封装修改数据的方法

原本的代码

from typing import Optional
from sqlmodel import Field, SQLModel, create_engine, Session, select, or_


class Hero(SQLModel, table=True):
    id: int | None = Field(default=None, primary_key=True)
    name: str = Field(index=True)
    secret_name: str
    age: int | None = Field(default=None, index=True)


# 创建数据库引擎
sqlite_url = "mysql+pymysql://root:root@127.0.0.1:3306/fastzdp_sqlmodel?charset=utf8mb4"
engine = create_engine(sqlite_url, echo=True)

# 创建所有表
SQLModel.metadata.drop_all(engine)
SQLModel.metadata.create_all(engine)

# 新增
hero_1 = Hero(name="Deadpond", secret_name="Dive Wilson")
hero_2 = Hero(name="Spider-Boy", secret_name="Pedro Parqueador")
hero_3 = Hero(name="Rusty-Man", secret_name="Tommy Sharp", age=48)
hero_4 = Hero(name="Tarantula", secret_name="Natalia Roman-on", age=32)
hero_5 = Hero(name="Black Lion", secret_name="Trevor Challa", age=35)
hero_6 = Hero(name="Dr. Weird", secret_name="Steve Weird", age=36)
hero_7 = Hero(name="Captain North America", secret_name="Esteban Rogelios", age=93)
with Session(engine) as session:
    session.add(hero_1)
    session.add(hero_2)
    session.add(hero_3)
    session.add(hero_4)
    session.add(hero_5)
    session.add(hero_6)
    session.add(hero_7)
    session.commit()

# 查询
with Session(engine) as session:
    statement = select(Hero).where(Hero.name == "Spider-Boy")
    results = session.exec(statement)
    hero = results.one()
    print("Hero:", hero)

    hero.age = 16
    session.add(hero)
    session.commit()
    session.refresh(hero)
    print("Updated hero:", hero)

简单的封装

def update(engine, model_obj, update_dict):
    """
    修改数据
    :param engine: 连接数据库的引擎对象
    :param model_obj: 模型对象
    :param update_dict: 更新字典
    :return:
    """
    with Session(engine) as session:
        if not isinstance(update_dict, dict):
            return
        for k, v in update_dict.items():
            if hasattr(model_obj, k):
                setattr(model_obj, k, v)
        session.add(model_obj)
        session.commit()
        session.refresh(model_obj)
        return model_obj

测试

from typing import Optional
from sqlmodel import Field, SQLModel
import fastzdp_sqlmodel as fsqlmodel


class User(SQLModel, table=True):
    id: Optional[int] = Field(default=None, primary_key=True)
    name: str
    age: Optional[int] = None


# 创建数据库引擎
engine = fsqlmodel.get_engine(database="fastzdp_sqlmodel")

# 初始化表格
fsqlmodel.init_table(engine)

# 创建数据
u = User(name="张三", age=23)
fsqlmodel.add(engine, u)

# 查询id为1的数据
u = fsqlmodel.get(engine, User, 1)
print(u)

# 修改
update_dict = {"name": "张三333"}
fsqlmodel.update(engine, u, update_dict)

# 查询id为1的数据
u = fsqlmodel.get(engine, User, 1)
print(u)

测试结果如下：
在这里插入图片描述

从测试结果来看是基本符合预期的。

本文来自互联网用户投稿，该文观点仅代表作者本人，不代表本站立场。本站仅提供信息存储空间服务，不拥有所有权，不承担相关法律责任。如若转载，请注明出处：http://www.coloradmin.cn/o/2047492.html

如若内容造成侵权/违法违规/事实不符，请联系多彩编程网进行投诉反馈，一经查实，立即删除！

相关文章

【笔记】PUML语法使用

【笔记】PUML语法使用

Sequence 时序图 类型&#xff1a;participant 1、排序 order participant NAME order NUMBER 默认是按出现定义的先后顺序排列&#xff0c;如下图。 【Sequence】未定义order的效果图 可以通过自定义改变&#xff0c;使用order关键字&#xff0c;将按ABC进行排序。 partici…
阅读更多...
20 以太网帧格式及命名方法

20 以太网帧格式及命名方法

以太网帧格式及命名方法 一、以太网帧格式及命名方法 &#xff08;一&#xff09;帧格式及命名标准 1.以太网帧格式 ​ 802.3 以太网帧格式 ​ 以太网帧数据的长度46-1500 ​ 以太网帧的长度64-1518 ​ 帧校验序列&#xff08;FCS&#xff09;&#xff1a;从目的地址开始…
阅读更多...
Binance 如何使用 Quickwit 构建 100PB 日志服务(Quickwit 博客)

Binance 如何使用 Quickwit 构建 100PB 日志服务(Quickwit 博客)

三年前&#xff0c;我们开源了 Quickwit&#xff0c;一个面向大规模数据集的分布式搜索引擎。我们的目标很宏大&#xff1a;创建一种全新的全文搜索引擎&#xff0c;其成本效率比 Elasticsearch 高十倍&#xff0c;配置和管理显著更简单&#xff0c;并且能够扩展到 PB 级别的数…
阅读更多...
【无标题】乐天HIQ壁挂炉使用

【无标题】乐天HIQ壁挂炉使用

这里写自定义目录标题 1.按键①&#xff1a; 按一下&#xff0c;小液晶显示的温度是所设定的供暖温度&#xff1b; 按二下&#xff0c;小液晶显示的温度是所设定的生活热水温度&#xff1b; 按三下&#xff0c;小液晶显示的温度是所设定的室内温度&#xff1b; 如果忘记按几下的…
阅读更多...
HCIE冲刺-----------论述解析

HCIE冲刺-----------论述解析

X园区 1.防火墙放行OSPF 2.trunk口阻塞vlan1 3.关闭VPN防环 4.ospf不通检查NSSA区域配置 5.重定向可以在三层口或二层口配置 6.vlan60在ospf的相关配置 Z园区 1.mpls不通检查lo0口配置 2.isis不通检查接口IP与lldp连接 3.确认同级RR还是备份RR 4.确认策略矩阵的访问 python 1…
阅读更多...
测试架构师要正直诚实而非率真

测试架构师要正直诚实而非率真

目录 正直诚实的重要性 为何需要正直诚实而非仅是率真 如何实践 正直不得罪人的方式 这里是一些关键的问题&#xff0c;在你感觉要“绝对诚实”地表达时考虑 率真这里有若干问题要考虑&#xff1a; 测试架构师作为软件开发团队中的关键角色之一&#xff0c;需要具备多种专…
阅读更多...
Python编码系列—Python SQL与NoSQL数据库交互：深入探索与实战应用

Python编码系列—Python SQL与NoSQL数据库交互：深入探索与实战应用

&#x1f31f;&#x1f31f; 欢迎来到我的技术小筑&#xff0c;一个专为技术探索者打造的交流空间。在这里&#xff0c;我们不仅分享代码的智慧&#xff0c;还探讨技术的深度与广度。无论您是资深开发者还是技术新手&#xff0c;这里都有一片属于您的天空。让我们在知识的海洋中…
阅读更多...
极狐GitLab 多行命令的 CI 日志管理体验的改进

极狐GitLab 多行命令的 CI 日志管理体验的改进

极狐GitLab 是 GitLab 在中国的发行版&#xff0c;专门面向中国程序员和企业提供企业级一体化 DevOps 平台&#xff0c;用来帮助用户实现需求管理、源代码托管、CI/CD、安全合规&#xff0c;而且所有的操作都是在一个平台上进行&#xff0c;省事省心省钱。可以一键安装极狐GitL…
阅读更多...
Python中15个递归函数经典案例解析

Python中15个递归函数经典案例解析

文末赠免费精品编程资料~~ 递归是Python编程中一个强大的工具&#xff0c;它允许函数调用自身以解决复杂问题。在本文中&#xff0c;我们将探索15个递归函数的经典案例&#xff0c;从基础到进阶&#xff0c;帮助你理解和掌握递归编程。 1. 阶乘计算 阶乘是一个常见的递归应用…
阅读更多...
CSP-CCF 202109-1 数组推导

CSP-CCF 202109-1 数组推导

一、问题描述 二、解答 初版&#xff1a; 只得了60分 #include<iostream> using namespace std; int main() {int n;cin >> n;int B[101] { 0 };int sum_max 0;int sum_min0;//以防错误“使用未初始化的局部变量”&#xff0c;建议所有变量都要初始化&#xf…
阅读更多...
2024 Google 开发者大会，沉浸式体验AI社会公益

2024 Google 开发者大会，沉浸式体验AI社会公益

文章目录 一、现场打卡二、AI 社会公益三、Gemma 模型四、Gemini 模型五、Google Cloud六、现场体验七、带着问题逛展八、学习资源和活动九、结束 Happy Hour 一、现场打卡 大家好&#xff0c;我是小雨。 2024 Google 开发者大会&#xff0c;沉浸式体验AI社会公益 今天我们参加…
阅读更多...
ERD Online即将突破 4500 用户的喜悦与展望

ERD Online即将突破 4500 用户的喜悦与展望

亲爱的朋友们&#xff1a; 大家好&#xff01; 今天&#xff0c;我怀着无比激动的心情&#xff0c;要与大家分享一个令人振奋的消息&#xff1a;我们的网站用户即将突破 4500 大关&#xff01; 这一路走来&#xff0c;充满了挑战与艰辛&#xff0c;但每一次的困难都成为了我们前…
阅读更多...
2024新型数字政府综合解决方案（十）

2024新型数字政府综合解决方案（十）

新型数字政府综合解决方案融合先进的人工智能、大数据、区块链及云计算技术&#xff0c;旨在通过数据共享、智能分析与自动化处理&#xff0c;打造高效、透明、安全的政务环境&#xff0c;优化服务流程&#xff0c;提升决策科学性&#xff0c;加强信息安全&#xff0c;实现政府…
阅读更多...
java入门-成员内部类和静态内部类的访问

java入门-成员内部类和静态内部类的访问

&#xff08;一&#xff09;成员内部类 package InnerClass;import javax.print.attribute.standard.MediaSize;public class Outer {//2外部类中的成员private int age99;public static String a;public class Inner{//普通的成员内部类//1.1成员变量public String name;priva…
阅读更多...
删除镜像报子镜像依赖错误

删除镜像报子镜像依赖错误

1、删除镜像报子镜像依赖错误 出现这个错误的原因是因为有其他镜像依赖需要删除的镜像。 2解决方法 2.1首先查看无法删除的镜像被哪些镜像所依赖 docker image inspect --format{{.RepoTags}} {{.Id}} {{.Parent}} $(docker image ls -q --filter since${image_id}) # ${ima…
阅读更多...
数据结构：线性结构之顺序表、链表篇

数据结构：线性结构之顺序表、链表篇

数据结构&#xff1a;顺序表、链表篇 线性表一、顺序表&#xff08;一&#xff09;顺序表的结构定义&#xff08;二&#xff09;顺序表的功能实现1、初始化2、销毁3、扩容4、插入5、删除 &#xff08;三&#xff09;顺序表例题分析1、删除有序数组中的重复项2、合并两个有序数组…
阅读更多...
【Hot100】LeetCode—73. 矩阵置零

【Hot100】LeetCode—73. 矩阵置零

目录 1- 思路开辟额外两个一维数组 2- 实现⭐53. 最大子数组和——题解思路 3- ACM 实现 原题连接&#xff1a;73. 矩阵置零 1- 思路 开辟额外两个一维数组 1- 利用额外的两个一维数组 boolean 数组空间&#xff0c;遇到 0 则将当前位置的元素设置为 true 一维 row 数组&…
阅读更多...
如何判断一个dll/exe是32位还是64位

如何判断一个dll/exe是32位还是64位

通过记事本判断&#xff08;可判断C或者C#&#xff09; 64位、将dll用记事本打开&#xff0c;可以看到一堆乱码&#xff0c;但是找到乱码行的第一个PE&#xff0c;如果后面是d?则为64位 32位、将dll用记事本打开&#xff0c;可以看到一堆乱码&#xff0c;但是找到乱码行的第…
阅读更多...
最好用的Linux发行版---WSL

最好用的Linux发行版---WSL

使用debian开发半年&#xff0c;那个号称稳定的操作系统&#xff0c;ubuntu也是基于它的testing版本开发的&#xff0c;在一次设置testing更新后英伟达驱动掉了、引导区无法启动、bios损坏&#xff0c;现在老实了&#xff0c;换回了Window&#xff0c;并且激进的选择了win11&am…
阅读更多...
c语言中的宏函数及c++的内联函数及auto及NULL

c语言中的宏函数及c++的内联函数及auto及NULL

c的内联函数 使用内联函数可以减少函数栈帧的开销。 Swap(a, b); 00A516C8 mov eax,dword ptr [a] 00A516CB mov dword ptr [ebp-20h],eax 00A516CE mov ecx,dword ptr [b] 00A516D1 mov dword ptr [a],ecx 00A516D4 mov …
阅读更多...
推荐文章
最新文章

Copyright @ 2022~2023