当前位置: 首页 > news >正文

python sqlite3 线程池封装

1. 封装 sqlite3

1.1. 依赖包引入

# -*- coding: utf-8 -*-
#import os
import sys
import datetime
import loggingimport sqlite3

1.2. 封装类

class SqliteTool(object):#def __init__(self, host, port, user, password, database):def __init__(self, host, database):self._host = host#self._port = port#self._user = user#self._password = passwordself._database = databaseself._pool = Noneself._maxconns = 6  # 连接池中最多有多少个连接print("__init__", self._database)

1.3. 连接池操作

    def init_pool(self):'''初始化连接池'''try:logging.info('Begin to create {0} postgresql pool on:{1}.\n'.format(self._host, datetime.datetime.now()))pool = []for _ in range(self._maxconns):# check_same_thread=False 支持多线程conn = sqlite3.connect(self._database, check_same_thread=False)pool.append(conn)self._pool = pool#print("init_pool", self._maxconns, len(self._pool), self._pool)logging.info('SUCCESS: create {0} postgresql pool success on {1}.\n'.format(self._host, datetime.datetime.now()))except Exception as e:logging.error('ERROR: create {0} postgresql pool failed on {1}.\n'.format(self._host, datetime.datetime.now()))self.close_pool()sys.exit('ERROR: create postgresql pool error caused by {0}'.format(str(e)))def close_pool(self):'''关闭 pool'''if self._pool != None:for conn in self._pool:conn.close()def get_conn(self):if not self._pool:self.init_pool()return self._pool.pop()def close_conn(self, conn):if self._pool:self._pool.append(conn)

1.4. 增删改查

1.4.1. 创建表

    # 创建数据表def create_table(self, sql: str):"""创建表:param sql: create sql语句:return: True表示创建表成功"""print("create_table", sql)result = Falsetry:conn = self.get_conn()cursor = conn.cursor()cursor.execute(sql)print("[create table success]")result = Trueexcept Exception as e:logging.error('ERROR: execute {0} causes error {1} in create_table'.format(sql, str(e)))sys.exit('ERROR: create table from database error caused {0}'.format(str(e)))finally:cursor.close()conn.commit()#conn.close()self.close_conn(conn)return result

1.4.2. 删除表

    # 删除数据表def drop_table(self, sql: str):"""删除表:param sql: drop sql语句:return: True表示删除成功"""result = Falsetry:conn = self.get_conn()cursor = conn.cursor()cursor.execute(sql)print("[drop table success]")result = Trueexcept Exception as e:logging.error('ERROR: execute {0} causes error {1} in drop_table'.format(sql, str(e)))sys.exit('ERROR: drop table from database error caused {0}'.format(str(e)))finally:cursor.close()conn.commit()#conn.close()self.close_conn(conn)return result

1.4.3. 插入数据

    def exec_insert(self, sql):'''执行插入'''result = Falsetry:conn = self.get_conn()cursor = conn.cursor()cursor.execute(sql)result =  Trueexcept Exception as e:logging.error('ERROR: execute {0} causes error {1} in exec_insert'.format(sql, str(e)))sys.exit('ERROR: insert data from database error caused {0}'.format(str(e)))finally:cursor.close()conn.commit()#conn.close()self.close_conn(conn)return resultdef exec_insert_plus(self, table: str, params: dict):'''执行插入'''result = Falsetry:key_tup = tuple(params.keys())key_str = ",".join(key_tup)# different with psqlval_str = ",".join(("?",)*len(key_tup))sql_str = "insert into " + table + " (" + key_str + ") values (" + val_str + ")"#val_tup = tuple(params.values())val_tup = ()for item in params.values():if type(item) == list:val_tup += (json.dumps(item),)elif type(item) == str:val_tup += (item,)elif type(item) == int:val_tup += (item,)else:val_tup += (item,)#val_tup.append(str(item))#print("exec_insert_plus", sql_str, val_tup)conn = self.get_conn()cursor = conn.cursor()cursor.execute(sql_str, val_tup)result =  Trueexcept Exception as e:logging.error('ERROR: execute {0} causes error {1} in exec_insert_plus'.format(table, str(e)))sys.exit('ERROR: insert data from database error caused {0}'.format(str(e)))finally:cursor.close()conn.commit()#conn.close()self.close_conn(conn)return resultdef exec_insert_many(self, table: str, params: List[Dict]):def dict_to_str(tab: str, param: Dict):key_tup = tuple(param.keys())key_str = ",".join(key_tup)# different with psqlval_str = ",".join(("?",)*len(key_tup))sql_str = "insert into " + tab + " (" + key_str + ") values (" + val_str + ")"return sql_strdef dict_to_tuple(param: Dict):val_tup = ()for item in param.values():if type(item) == list:val_tup += (json.dumps(item),)elif type(item) == str:val_tup += (item,)elif type(item) == int:val_tup += (item,)else:val_tup += (item,)#val_tup.append(str(item))return val_tup'''执行插入'''result = Falseif len(params) <= 0:return resulttry:sql_str = dict_to_str(table, params[0])val_lst = []for param in params:val_lst.append(dict_to_tuple(param))print("exec_insert_many", sql_str, val_lst)conn = self.get_conn()cursor = conn.cursor()cursor.executemany(sql_str, val_lst)result =  Trueexcept Exception as e:logging.error('ERROR: execute {0} causes error {1} in exec_insert'.format(sql_str, str(e)))sys.exit('ERROR: insert data from database error caused {0}'.format(str(e)))finally:cursor.close()conn.commit()#conn.close()self.close_conn(conn)return result

1.4.4. 删除数据

	# sql = "DELETE from users where user_id='83f7d86b594e4b26a7196ab761afcc7c';"def exec_delete(self, sql):'''执行查询'''result = Falsetry:conn = self.get_conn()cursor = conn.cursor()cursor.execute(sql)result =  Trueexcept Exception as e:logging.error('ERROR: execute {0} causes error {1} in exec_delete'.format(sql, str(e)))sys.exit('ERROR: delete data from database error caused {0}'.format(str(e)))finally:cursor.close()conn.commit()#conn.close()self.close_conn(conn)return result

1.4.5. 更新数据

	# 修改单个值# update tasks set status='running' where task_id='0791216839b04d5c88846817f78280cc';# 修改多个值# update tasks set status='running',score='10' where task_id='0791216839b04d5c88846817f78280cc';def exec_update(self, sql):'''执行更新'''result = Falsetry:conn = self.get_conn()cursor = conn.cursor()cursor.execute(sql)result =  Trueexcept Exception as e:logging.error('ERROR: execute {0} causes error {1} in exec_update'.format(sql, str(e)))sys.exit('ERROR: update data from database error caused {0}'.format(str(e)))finally:cursor.close()conn.commit()#conn.close()self.close_conn(conn)return result

1.4.6. 查询数据

    # select * from users where user_name='hello';def exec_select(self, sql):'''执行查询'''try:conn = self.get_conn()cursor = conn.cursor()cursor.execute(sql)result = cursor.fetchall()#result = cursor.fetchone()except Exception as e:logging.error('ERROR: execute {0} causes error {1} in exec_select'.format(sql, str(e)))sys.exit('ERROR: load data from database error caused {0}'.format(str(e)))finally:cursor.close()#conn.close()#print("exec_select", len(self._pool), self._pool)print("init_pool", self._maxconns, len(self._pool), self._pool)self.close_conn(conn)return result

1.4.7. 测试

    def test_select(self, sql):result = self.exec_select(sql)print("test_select", result)return result

2. 操作使用实例

# for test
from typing import Optional, List, Dict, Union
from pydantic import BaseModel, Fieldclass ......if __name__ == '__main__':dbhost = ""dbdatabase = "./test.db"db = SqliteTool(dbhost, dbdatabase)class TaskInDB(BaseModel):task_id: strdisabled: intdef create_tests_table(db):sql_str = "create table if not exists tests("sql_str += "task_id char(32) primary key,"sql_str += "disabled int not null"sql_str += ");"return db.create_table(sql_str)def drop_tests_table(db):sql_str = "drop table if exists tests;"return db.drop_table(sql_str)def get_tests_indb(db, tasks: Union[List[str], str, None] = None):if tasks == None:sql_str = "select * from tests;"if type(tasks) == list:print("list", tasks)key_str = ",".join(tasks)sql_str = "select * from tests where task_id in (" + key_str + ");"elif type(tasks) == str:print("str", tasks)sql_str = "select * from tests where task_id ='"+tasks+"';"elif type(tasks) == None:print("none")ret_tasks = db.exec_select(sql_str)return ret_tasksdef create_tests_indb(db, tasks: List[TaskInDB]):#return db.exec_insert_plus("tests", task_indb.model_dump())#return db.exec_insert_plus("tests", task_indb.dict())params = []for task in tasks:print("create_tests_indb", task)#params.append(task.model_dump())params.append(task.dict())return db.exec_insert_many("tests", params)#  重建数据库if not drop_tests_table(db) or not create_tests_table(db):print("ERROR")# 创建两条记录task_indb1 = TaskInDB(task_id="11111111", disabled=1)#create_tests_indb(db, task_indb)task_indb2 = TaskInDB(task_id="22222222", disabled=0)#create_tests_indb(db, task_indb)task_indb3 = TaskInDB(task_id="33333333", disabled=0)#create_tests_indb(db, task_indb)create_tests_indb(db, [task_indb1, task_indb2, task_indb3])# 查询记录#key_tup = tuple(TaskInDB.model_fields.keys())key_tup = tuple(TaskInDB.__fields__.keys())#key_str = ",".join(key_tup)# allret_tasks = get_tests_indb(db)#  str#ret_tasks = get_tests_indb(db, tasks="11111111")# list#ret_tasks = get_tests_indb(db, tasks=["11111111", "22222222"])for ret_task in ret_tasks:print(ret_task)task_indb = TaskInDB(**{key: ret_task[i] for i,key in enumerate(key_tup)})print(task_indb)print("OK")

相关文章:

python sqlite3 线程池封装

1. 封装 sqlite3 1.1. 依赖包引入 # -*- coding: utf-8 -*- #import os import sys import datetime import loggingimport sqlite31.2. 封装类 class SqliteTool(object):#def __init__(self, host, port, user, password, database):def __init__(self, host, database):s…...

亚马逊运营:如何通过自养号测评有效防关联,避免砍单

店铺安全对于跨境电商卖家至关重要&#xff0c;它是我们业务稳定运营的基础。一旦店铺遭到亚马逊的封禁&#xff0c;往往意味着巨大的损失。因此&#xff0c;合规运营已经成为了卖家们的共识。然而&#xff0c;许多卖家可能会因为一些看似微小的失误&#xff0c;导致店铺被关联…...

winfrom图像加速渲染时图像不显示

winform中加入这段代码&#xff0c;即使不调用也会起作用&#xff1b;当图像不显示时&#xff0c;可以注释掉这段代码...

Redash 默认key漏洞(CVE-2021-41192)复现

Redash是以色列Redash公司的一套数据整合分析解决方案。该产品支持数据整合、数据可视化、查询编辑和数据共享等。 Redash 10.0.0及之前版本存在安全漏洞&#xff0c;攻击者可利用该漏洞来使用已知的默认值伪造会话。 1.漏洞级别 中危 2.漏洞搜索 fofa "redash"…...

Git学习笔记:3 git tag命令

文章目录 git tag 基本用法1. 创建标签2. 查看标签3. 删除标签4. 推送标签到远程仓库5. 检出标签 普通提交和标签的区别1. 提交&#xff08;Commit&#xff09;2. 标签&#xff08;Tag&#xff09; git tag 基本用法 git tag 是 Git 中用于管理和操作标签&#xff08;tag&…...

10年软件测试经验,该有什么新的职业规划?

个人觉得&#xff0c;最关键是识别个人的兴趣和长期目标&#xff0c;以及市场需求&#xff0c;制定符合自己职业发展的规划&#xff0c;列了几个常见的方向&#xff1a; 1. 技术深化 专业领域专长&#xff1a;在某一测试领域&#xff08;如自动化测试、性能测试、安全测试等&am…...

重构改善既有代码的设计-学习(四):简化条件逻辑

1、分解条件表达式&#xff08;Decompose Conditional&#xff09; 可以将大块代码分解为多个独立的函数&#xff0c;根据每个小块代码的用途&#xff0c;为分解而得的新函数命名。对于条件逻辑&#xff0c;将每个分支条件分解成新函数还可以带来更多好处&#xff1a;可以突出条…...

【代码---利用一个小程序,读取文件夹中图片,将其合成为一个视频】

提示&#xff1a;文章写完后&#xff0c;目录可以自动生成&#xff0c;如何生成可参考右边的帮助文档 文章目录 前言程序详细说明总结 前言 提示&#xff1a;这里可以添加本文要记录的大概内容&#xff1a; 创建一个程序将图像合成为视频通常需要使用图像处理和视频编码库。 …...

MVC 和 MVVM的区别

MVC&#xff1a; M&#xff08;model数据&#xff09;、V&#xff08;view视图&#xff09;&#xff0c;C&#xff08;controlle控制器&#xff09; 缺点是前后端无法独立开发&#xff0c;必须等后端接口做好了才可以往下走&#xff1b; 前端没有自己的数据中心&#xff0c;太…...

redis—Set集合

目录 前言 1.常见命令 2.使用场景 前言 集合类型也是保存多个字符串类型的元素的&#xff0c;但和列表类型不同的是&#xff0c;集合中1)元素之间是无序的2)元素不允许重复&#xff0c;如图2-24所示。一个集合中最多可以存储22 - 1个元素。Redis 除了支持集合内的增删查改操…...

【jetson笔记】vscode远程调试

vscode安装插件 vscode安装远程插件Remote-SSH 安装完毕点击左侧远程资源管理器 打开SSH配置文件 添加如下内容&#xff0c;Hostname为jetson IP&#xff0c;User为登录用户名需替换为自己的 Host aliasHostName 192.168.219.57User jetson配置好点击连接&#xff0c;控制台输…...

大数据处理流程包括哪些环节

大数据处理流程作为当今信息时代的关键技术之一&#xff0c;已经成为各个行业的必备工具。这个流程涵盖了从数据收集、存储、处理、分析到应用的各个环节&#xff0c;确保了数据的有效利用和价值的最大化。 一、数据收集 随着物联网、移动互联网、社交媒体等领域的快速发展&a…...

C++入门篇章1(C++是如何解决C语言不能解决的问题的)

目录 1.C关键字(以C98为例)2.命名空间2.1 命名空间定义2.2命名空间使用 3.C输入&输出4.缺省参数4.1缺省参数概念4.2 缺省参数分类 5. 函数重载5.1函数重载概念5.2 C支持函数重载的原理--名字修饰(name Mangling) 1.C关键字(以C98为例) C总计63个关键字&#xff0c;C语言32…...

java复习篇 数据结构:链表第一节

目录 单向链表 初始 头插 思路 情况一 情况二 代码 尾插 思路 遍历 优化遍历 遍历验证头插 尾插代码 优化 尾插测试 get 思路 代码 测试 insert 思路 代码 优化 测试 remove 移除头结点 提问 移除指定位置 测试 单向链表 每个元素只知道自己的下一个…...

深入理解与运用Lombok的@Cleanup注解:自动化资源管理利器

前言 在Java编程中&#xff0c;正确地管理和释放诸如文件流、数据库连接等资源至关重要。若处理不当&#xff0c;可能会引发内存泄漏或系统资源耗尽等问题。为此&#xff0c;Lombok库提供了一个名为Cleanup的便捷注解&#xff0c;它允许我们以简洁且安全的方式自动关闭实现了j…...

【LeetCode每日一题】2865. 美丽塔 I

2024-1-24 文章目录 [2865. 美丽塔 I](https://leetcode.cn/problems/beautiful-towers-i/) 2865. 美丽塔 I 初始化变量 ans 为0&#xff0c;用于记录最大的和值。获取整数列表的长度&#xff0c;保存到变量 n 中。使用一个循环遍历列表中的每个位置&#xff0c;从0到n-1。在循…...

Cute Http File Server 使用文章

下载 官网&#xff1a;http://iscute.cn/chfs 蓝奏下载&#xff1a;https://wwts.lanpw.com/iKP1i1m9572h 开源&#xff1a;https://github.com/docblue/chfsgui 介绍 Cute Http File Server 是国内免费开源的局域网传输服务器软件。 可以不用借助QQ、某信软件传输文件&am…...

c#算法(10)——求点到直线的距离

前言 在上位机软件开发领域,特别是机器视觉领域,经常会遇到尺寸测量的场景,比如让我们求一个点到一条直线的距离,我们已知了直线上的两个点的坐标,然后又已知了直线外的一个点的坐标,那么如何求出该直线外的一点到直线的距离呢?本文就是来讲解如何求点到直线的距离的,…...

[小脚本] maya 命令行常用操作

其实这些代码大部分是从 chatgpt 中生成的。 骨骼命名 import maya.cmds as cmdsdef rename_bones():selected_bones cmds.ls(type"joint") # 获取选中的骨骼for bone in selected_bones:if "_" in bone:new_name bone.split("_")[0] # 获…...

数据结构·单链表

不可否认的是&#xff0c;前几节我们讲解的顺序表存在一下几点问题&#xff1a; 1. 中间、头部的插入和删除&#xff0c;需要移动一整串数据&#xff0c;时间复杂度O(N) 2. 增容需要申请新空间&#xff0c;拷贝数据&#xff0c;释放旧空间。会有不小的消耗 3. 增容一般是2倍的增…...

wordpress后台更新后 前端没变化的解决方法

使用siteground主机的wordpress网站&#xff0c;会出现更新了网站内容和修改了php模板文件、js文件、css文件、图片文件后&#xff0c;网站没有变化的情况。 不熟悉siteground主机的新手&#xff0c;遇到这个问题&#xff0c;就很抓狂&#xff0c;明明是哪都没操作错误&#x…...

Chapter03-Authentication vulnerabilities

文章目录 1. 身份验证简介1.1 What is authentication1.2 difference between authentication and authorization1.3 身份验证机制失效的原因1.4 身份验证机制失效的影响 2. 基于登录功能的漏洞2.1 密码爆破2.2 用户名枚举2.3 有缺陷的暴力破解防护2.3.1 如果用户登录尝试失败次…...

stm32G473的flash模式是单bank还是双bank?

今天突然有人stm32G473的flash模式是单bank还是双bank&#xff1f;由于时间太久&#xff0c;我真忘记了。搜搜发现&#xff0c;还真有人和我一样。见下面的链接&#xff1a;https://shequ.stmicroelectronics.cn/forum.php?modviewthread&tid644563 根据STM32G4系列参考手…...

大数据零基础学习day1之环境准备和大数据初步理解

学习大数据会使用到多台Linux服务器。 一、环境准备 1、VMware 基于VMware构建Linux虚拟机 是大数据从业者或者IT从业者的必备技能之一也是成本低廉的方案 所以VMware虚拟机方案是必须要学习的。 &#xff08;1&#xff09;设置网关 打开VMware虚拟机&#xff0c;点击编辑…...

转转集团旗下首家二手多品类循环仓店“超级转转”开业

6月9日&#xff0c;国内领先的循环经济企业转转集团旗下首家二手多品类循环仓店“超级转转”正式开业。 转转集团创始人兼CEO黄炜、转转循环时尚发起人朱珠、转转集团COO兼红布林CEO胡伟琨、王府井集团副总裁祝捷等出席了开业剪彩仪式。 据「TMT星球」了解&#xff0c;“超级…...

基础测试工具使用经验

背景 vtune&#xff0c;perf, nsight system等基础测试工具&#xff0c;都是用过的&#xff0c;但是没有记录&#xff0c;都逐渐忘了。所以写这篇博客总结记录一下&#xff0c;只要以后发现新的用法&#xff0c;就记得来编辑补充一下 perf 比较基础的用法&#xff1a; 先改这…...

高等数学(下)题型笔记(八)空间解析几何与向量代数

目录 0 前言 1 向量的点乘 1.1 基本公式 1.2 例题 2 向量的叉乘 2.1 基础知识 2.2 例题 3 空间平面方程 3.1 基础知识 3.2 例题 4 空间直线方程 4.1 基础知识 4.2 例题 5 旋转曲面及其方程 5.1 基础知识 5.2 例题 6 空间曲面的法线与切平面 6.1 基础知识 6.2…...

苍穹外卖--缓存菜品

1.问题说明 用户端小程序展示的菜品数据都是通过查询数据库获得&#xff0c;如果用户端访问量比较大&#xff0c;数据库访问压力随之增大 2.实现思路 通过Redis来缓存菜品数据&#xff0c;减少数据库查询操作。 缓存逻辑分析&#xff1a; ①每个分类下的菜品保持一份缓存数据…...

CocosCreator 之 JavaScript/TypeScript和Java的相互交互

引擎版本&#xff1a; 3.8.1 语言&#xff1a; JavaScript/TypeScript、C、Java 环境&#xff1a;Window 参考&#xff1a;Java原生反射机制 您好&#xff0c;我是鹤九日&#xff01; 回顾 在上篇文章中&#xff1a;CocosCreator Android项目接入UnityAds 广告SDK。 我们简单讲…...

Java线上CPU飙高问题排查全指南

一、引言 在Java应用的线上运行环境中&#xff0c;CPU飙高是一个常见且棘手的性能问题。当系统出现CPU飙高时&#xff0c;通常会导致应用响应缓慢&#xff0c;甚至服务不可用&#xff0c;严重影响用户体验和业务运行。因此&#xff0c;掌握一套科学有效的CPU飙高问题排查方法&…...