psycopg2 使用dbutils 工具封装

1.什么是dbutils

Dbutils是一套工具,可为数据库提供可靠,持久和汇总的连接,该连接可在各种多线程环境中使用。

2.使用代码记录

db_config.py  数据库配置类:

# -*- coding: UTF-8 -*-
import psycopg2# 数据库信息
DB_TEST_HOST = "dev-pg.test.xxx.cloud"
DB_TEST_PORT = 1921
DB_TEST_DBNAME = "check_db"
DB_TEST_USER = "check_db"
DB_TEST_PASSWORD = "okk3py2323gBaX111"# 数据库连接编码
DB_CHARSET = "utf8"# mincached : 启动时开启的闲置连接数量(缺省值 0 开始时不创建连接)
DB_MIN_CACHED = 10# maxcached : 连接池中允许的闲置的最多连接数量(缺省值 0 代表不闲置连接池大小)
DB_MAX_CACHED = 10# maxshared : 共享连接数允许的最大数量(缺省值 0 代表所有连接都是专用的)如果达到了最大数量,被请求为共享的连接将会被共享使用
DB_MAX_SHARED = 20# maxconnecyions : 创建连接池的最大数量(缺省值 0 代表不限制)
DB_MAX_CONNECYIONS = 100# blocking : 设置在连接池达到最大数量时的行为(缺省值 0 或 False 代表返回一个错误<toMany......> 其他代表阻塞直到连接数减少,连接被分配)
DB_BLOCKING = True# maxusage : 单个连接的最大允许复用次数(缺省值 0 或 False 代表不限制的复用).当达到最大数时,连接会自动重新连接(关闭和重新打开)
DB_MAX_USAGE = 0# setsession : 一个可选的SQL命令列表用于准备每个会话,如["set datestyle to german", ...]
DB_SET_SESSION = None# creator : 使用连接数据库的模块
DB_CREATOR = psycopg2

2.封装sqlHelper.py

import sys
import db_config as config
import psycopg2.extras
from dbutils.pooled_db import PooledDB
import threadingclass PsycopgConn:_instance_lock = threading.Lock()def __init__(self):self.init_pool()def __new__(cls, *args, **kwargs):if not hasattr(cls, '_instance'):with PsycopgConn._instance_lock:if not hasattr(cls, '_instance'):PsycopgConn._instance = object.__new__(cls)return PsycopgConn._instancedef get_pool_conn(self):"""获取连接池连接:return:"""if not self._pool:self.init_pool()return self._pool.connection()def init_pool(self):"""初始化连接池:return:"""try:pool = PooledDB(creator=config.DB_CREATOR,  # 使用连接数据库的模块 psycopg2maxconnections=config.DB_MAX_CONNECYIONS,  # 连接池允许的最大连接数,0 和 None 表示不限制连接数mincached=config.DB_MIN_CACHED,  # 初始化时,链接池中至少创建的空闲的链接,0 表示不创建maxcached=config.DB_MAX_CACHED,  # 链接池中最多闲置的链接,0 和 None 不限制blocking=True,  # 连接池中如果没有可用连接后,是否阻塞等待。True,等待;False,不等待然后报错maxusage=None,  # 一个链接最多被重复使用的次数,None 表示无限制setsession=[],  # 开始会话前执行的命令列表host=config.DB_TEST_HOST,port=config.DB_TEST_PORT,user=config.DB_TEST_USER,password=config.DB_TEST_PASSWORD,database=config.DB_TEST_DBNAME)self._pool = poolexcept:print'connect postgresql error'self.close_pool()def close_pool(self):"""关闭连接池连接:return:"""if self._pool != None:self._pool.close()def SelectSql(self, sql):"""查询:param sql::return:"""try:conn = self.get_pool_conn()cursor = conn.cursor(cursor_factory=psycopg2.extras.RealDictCursor)  # 设置返回格式为字典cursor.execute(sql)result = cursor.fetchall()except Exception as e:print('execute sql {0} is error'.format(sql))sys.exit('ERROR: load data from database error caused {0}'.format(str(e)))finally:cursor.close()conn.close()return resultdef InsertSql(self, sql):"""插入数据:param sql::return:"""try:conn = self.get_pool_conn()cursor = conn.cursor()cursor.execute(sql)result = Trueexcept Exception as e:print('ERROR: execute  {0} causes error'.format(sql))sys.exit('ERROR: update data from database error caused {0}'.format(str(e)))finally:cursor.close()conn.commit()conn.close()return resultdef UpdateSql(self, sql):"""更新数据:param sql::return:"""try:conn = self.get_pool_conn()cursor = conn.cursor()cursor.execute(sql)result = Trueexcept Exception as e:print('ERROR: execute  {0} causes error'.format(sql))sys.exit('ERROR: update data from database error caused {0}'.format(str(e)))finally:cursor.close()conn.commit()conn.close()return resultif __name__ == "__main__":pgsql = PsycopgConn()# 打开一个文件with open('task.txt') as fr:# 读取文件所有行lines = fr.readlines()lines = [i.rstrip() for i in lines]list = []list.append("taskId,batchId\n")for taskId in lines:sql = "select task_id, batch_id  from ics_batch_info where batch_id=(select MAX(batch_id) from ics_batch_info WHERE task_id = '{taskId}')".format(taskId=taskId)print("执行sql: ", sql)result = pgsql.SelectSql(sql)if len(result) != 0:list.append(result[0] + "," + str(result[1]) + "\n")print(result)list[len(list) - 1] = list[len(list) - 1].rstrip();with open("最大批次查询结果.csv", 'w') as fw:fw.writelines(list)print("☺☺☺执行完毕☺☺☺")

验证执行结果:

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

如若内容造成侵权/违法违规/事实不符,请联系多彩编程网进行投诉反馈email:809451989@qq.com,一经查实,立即删除!

相关文章

docker 安装 redis

目录 1、下载镜像文件 2、创建实例并启动 3、使用 redis 镜像执行 redis-cli 命令连接 4、redis持久化操作 5、然后按照第3点&#xff0c;再试一试&#xff0c;看看redis持久化是否配置成功。 6、最后与redis可视化工具测试连接 大家先 su root&#xff0c;这让输入命令就…

GEE/PIE遥感大数据处理与典型案例

随着航空、航天、近地空间等多个遥感平台的不断发展&#xff0c;近年来遥感技术突飞猛进。由此&#xff0c;遥感数据的空间、时间、光谱分辨率不断提高&#xff0c;数据量也大幅增长&#xff0c;使其越来越具有大数据特征。对于相关研究而言&#xff0c;遥感大数据的出现为其提…

23种设计模式攻关

&#x1f44d;一、创建者模式 &#x1f516;1.1、单例模式 单例模式&#xff08;Singleton Pattern&#xff09;&#xff0c;用于确保一个类只有一个实例&#xff0c;并提供全局访问点。 在某些情况下&#xff0c;我们需要确保一个类只能有一个实例&#xff0c;比如数据库连接…

LSTM模型

目录 LSTM模型 LSTM结构图 LSTM的核心思想 细胞状态 遗忘门 输入门 输出门 RNN模型 LRNN LSTM模型 什么是LSTM模型 LSTM (Long Short-Term Memory)也称长短时记忆结构,它是传统RNN的变体,与经典RNN相比能够有效捕捉长序列之间的语义关联,缓解梯度消失或爆炸现象.同时LS…

SpringCloud学习笔记(四)_ZooKeeper注册中心

基于Spring Cloud实现服务的发布与调用。而在18年7月份&#xff0c;Eureka2.0宣布停更了&#xff0c;将不再进行开发&#xff0c;所以对于公司技术选型来说&#xff0c;可能会换用其他方案做注册中心。本章学习便是使用ZooKeeper作为注册中心。 本章使用的zookeeper版本是 3.6…

SpringCloud 教程 | 第一篇: 服务的注册与发现(Eureka)

一、spring cloud简介 spring cloud 为开发人员提供了快速构建分布式系统的一些工具&#xff0c;包括配置管理、服务发现、断路器、路由、微代理、事件总线、全局锁、决策竞选、分布式会话等等。它运行环境简单&#xff0c;可以在开发人员的电脑上跑。另外说明spring cloud是基…

国产精品:讯飞星火最新大模型V2.0

大家好&#xff0c;我是爱编程的喵喵。双985硕士毕业&#xff0c;现担任全栈工程师一职&#xff0c;热衷于将数据思维应用到工作与生活中。从事机器学习以及相关的前后端开发工作。曾在阿里云、科大讯飞、CCF等比赛获得多次Top名次。现为CSDN博客专家、人工智能领域优质创作者。…

评测凯迪仕K70「千里眼」智能锁:不忘安全初心,便捷体验更上一层

能打败凯迪仕的&#xff0c;只有它自己。这是我们在体验过凯迪仕最新旗舰产品K70「千里眼」智能锁之后的感受。作为凯迪仕2023年最新旗舰机型&#xff0c;K70「千里眼」智能锁在配置上可以说是「机皇」般的存在。3K超高清智能锁猫眼、车规级24GHz雷达、大小双屏设计、三方可视对…

百度云BOS云存储的图片如何在访问时,同时进行格式转换、缩放等处理

前言 之前做了一个图片格式转换和压缩的服务&#xff0c;结果太占内存。后来查到在访问图片链接时&#xff0c;支持进行图片压缩和格式转换&#xff0c;本来想着先格式转换、压缩图片再上传到BOS&#xff0c;现在变成了上传后&#xff0c;访问时进行压缩和格式转换。想了想&am…

Jobs Portal求职招聘系统源码v3.5版本

Jobs Portal求职招聘系统 是为求职者和公司发布职位而开发的交互式求职招聘源码。它使求职者能够发布简历、搜索工作、查看个人工作列表。 它将提供各种公司在网站上放置他们的职位空缺资料&#xff0c;并且还可以选择搜索候选人简历。 除此之外&#xff0c;还有一个管理模块供…

基于AVR128单片机世界电子时钟的设计

一、系统方案 上电初始化完成系统初始化&#xff0c;液晶滚动显示北京、莫斯科、东京、伦敦、巴黎、纽约等六个城市的标准时间&#xff0c;显示的内容包括地区名及相应地区的年、月、日、星期、时、分、秒。 使用K1按键控制滚动显示或稳定显示某个地区的时间。 使用K3、K4、K5按…

Q/GDW 1597-2015《国家电网公司应用软件系统通用安全要求》

电力安全测试报告 电力行业检测标准 随着信息技术的快速发展和广泛应用&#xff0c;应用软件系统已成为企业信息化建设中不可或缺的重要组成部分。然而&#xff0c;应用软件系统的安全问题也随之而来&#xff0c;给企业和用户带来了潜在的风险和威胁。为了提高应用软件系统的…

信息与通信工程面试准备——信号与系统|10:23

8月16日 23:21 目录 ​编辑 1. 调制的作用 2. 放大器与振荡器的作用和区别 工作原理 输出信号 应用 反馈方式 设计复杂度 装置性质 3. 信号与系统&#xff1a;三大变换之间的关系&#xff1f; 4. 无码间串扰的条件 5. 冲激函数的作用&#xff1f; 研究的意义&…

Docker容器:docker数据管理、镜像的创建及dockerfile案例

文章目录 一、docker数据管理1.为何需要docker数据管理2.数据管理类型3.数据卷4.数据卷容器5.容器的互联 二.docker镜像的三种创建方法1.基于现有镜像创建1.1 启动镜像1.2 生成新镜像 2.基于本地模板创建2.1 OPENVZ 下载模板2.2 导入容器生成镜像 3.基于dockerfile创建3.1 dock…

Code Lab - 2

pip install torch-scatter -f https://pytorch-geometric.com/whl/torch-1.10.2cu102.html pip install torch-sparse -f https://pytorch-geometric.com/whl/torch-1.10.2cu102.html pip install torch-geometric pip install ogb 1. PyG Datasets PyG有两个类&#xff0c;用…

【Unity每日一记】SceneManager场景资源动态加载

&#x1f468;‍&#x1f4bb;个人主页&#xff1a;元宇宙-秩沅 &#x1f468;‍&#x1f4bb; hallo 欢迎 点赞&#x1f44d; 收藏⭐ 留言&#x1f4dd; 加关注✅! &#x1f468;‍&#x1f4bb; 本文由 秩沅 原创 &#x1f468;‍&#x1f4bb; 收录于专栏&#xff1a;uni…

elementui 修改日期选择器el-date-picker样式

1. 案例&#xff1a; 2. css /* 最外层颜色 */ .el-popper.is-pure {background: url("/assets/imgList/memuBG.png") no-repeat;border: none;background-size:100% 100%}/* 日期 1.背景透明 */ .el-date-picker{background: transparent; }/* 日期 2.标题、左右图…

关于css 的选择器和 css变量

css 选择器 常用的选择器 1. 后代选择器&#xff1a;也就是我们常见的空格选择器&#xff0c;选择的对象为该元素下的所有子元素 。例如&#xff0c;选择所有 元素下的 元素 div p{font-size:14px}2. 子元素选择器 ‘>’ 选择某元素下的直接子元素。例如&#xff0c;选择所…

Java算法_ BST 中第 k 个最小元素 (LeetCode_Hot100)

题目描述&#xff1a;给定一个二叉搜索树的根节点 &#xff0c;和一个整数 &#xff0c;请你设计一个算法查找其中第 个最小元素&#xff08;从 1 开始计数&#xff09;。 获得更多&#xff1f;算法思路:代码文档&#xff0c;算法解析的私得。 运行效果 完整代码 /*** 2 * Aut…

LeetCode150道面试经典题-- 二叉树的最大深度(简单)

1.题目 给定一个二叉树 root &#xff0c;返回其最大深度。 二叉树的 最大深度 是指从根节点到最远叶子节点的最长路径上的节点数。 2.示例 3.思路 深度优先遍历 一个二叉树要查询到最大深度&#xff0c;可以将问题转为从根节点出发&#xff0c;查看左右子树的最大深度&am…