营销型网站 平台,wordpress调用自定义类型文章,企业年金个人查询官网,织梦电影网站免费模板提示#xff1a;文章写完后#xff0c;目录可以自动生成#xff0c;如何生成可参考右边的帮助文档 文章目录 前言一、mysql连接池#xff1f;二、使用步骤1.引入库 前言
提示#xff1a;这里可以添加本文要记录的大概内容#xff1a;
例如#xff1a; 提示#xff1a… 提示文章写完后目录可以自动生成如何生成可参考右边的帮助文档 文章目录 前言一、mysql连接池二、使用步骤1.引入库 前言
提示这里可以添加本文要记录的大概内容
例如 提示以下是本篇文章正文内容下面案例可供参考
一、mysql连接池
安装包 DBUtils pip install DBUtils1.3
二、使用步骤
1.引入库
代码如下示例
# -*- coding:utf-8 -*-
# author: cai bao jun
# datetime: 2024/3/1 11:38
# File: 4数据库操作2.pyimport pymysql
from DBUtils.PooledDB import PooledDB
import datetimefrom logger import logger#### DBUtils 1.3
#### DBUtils 1.3
#### DBUtils 1.3class MysqlConfig(object):database test2022 # 测试 trainerNhost 127.0.0.1user rootport 3306password root# Mysql数据库相关操作
# Singleton
class DMLMysql(object):_pool None_isinstance None_flag Truedef __new__(cls, *args, **kwargs):if not cls._isinstance:print(new)cls._pool PooledDB(creatorpymysql, # 使用链接数据库的模块mincached10, # 初始化时链接池中至少创建的链接0表示不创建maxconnections200, # 连接池允许的最大连接数0和None表示不限制连接数blockingTrue, # 连接池中如果没有可用连接后是否阻塞等待。True等待False不等待然后报错hostMysqlConfig.host,portMysqlConfig.port,userMysqlConfig.user,passwordMysqlConfig.password,databaseMysqlConfig.database,)cls._isinstance super().__new__(cls)return cls._isinstancedef __init__(self, hostMysqlConfig.host, databaseMysqlConfig.database, userMysqlConfig.user, passwordMysqlConfig.password, portMysqlConfig.port):try:# print(开始链接mysql22332)self.database databaseself.pool DMLMysql._poolexcept Exception as e:logger.error(fdatabase connect error message is {str(e)})passpassdef open(self):self.conn self.pool.connection()self.cursor self.conn.cursor() # 表示读取的数据为字典类型return self.conn, self.cursordef close(self, cursor, conn):cursor.close()conn.close()def execute_sql(self, sqlQuery, value)::param sqlQuery: 拼接好的sql语句:param value: 需要拼接的值:return:try:conn, cursor self.open()conn.ping(reconnectTrue) # 超时断开重连cursor.execute(sqlQuery, value)# logger.info(数据执行成功)except Exception as e:logger.error(fdatabase name is {self.database} error info is{str(e)},sql is : {sqlQuery})conn.rollback()else:conn.commit()finally:self.close(cursor, conn)def select_sql(self,sqlQuery, value):ret Nonetry:conn, cursor self.open()conn.ping(reconnectTrue) # 超时断开重连cursor.execute(sqlQuery, value)ret cursor.fetchall()# logger.info(查询数据执行成功)except Exception as e:logger.error(fdatabase name is {self.database} error info is{str(e)},sql is : {sqlQuery})# self.conn.rollback()finally:self.close(cursor, conn)return retdef __del__(self):# self.cursor.close()# self.conn.close()# print(关闭mysql22332)passif __name__ __main__:dml DMLMysql()select_sql select author_id,category_id,views from article where id%svalue (1,)ret1 dml.select_sql(sqlQueryselect_sql,valuevalue)print(ret1)dml1 DMLMysql()dml2 DMLMysql()print(id(dml1))print(id(dml2))print(id(dml1)id(dml2))pass
