import pymysql.cursors
from dbutils.pooled_db import PooledDB
host = '127.0.0.1'
port = 3306
user = 'root'
password = 'root'
database = 'report'
class MySQLConnectionPool:
def __init__(self, ):
self.pool = PooledDB(
creator=pymysql, # 使用链接数据库的模块
mincached=10, # 初始化时,链接池中至少创建的链接,0表示不创建
maxconnections=200, # 连接池允许的最大连接数,0和None表示不限制连接数
blocking=True, # 连接池中如果没有可用连接后,是否阻塞等待。True,等待;False,不等待然后报错
host=host,
port=port,
user=user,
password=password,
database=database
)
def open(self):
self.conn = self.pool.connection()
self.cursor = self.conn.cursor(cursor=pymysql.cursors.DictCursor) # 表示读取的数据为字典类型
return self.conn, self.cursor
def close(self, cursor, conn):
cursor.close()
conn.close()
def select_one(self, sql, *args):
"""查询单条数据"""
conn, cursor = self.open()
cursor.execute(sql, args)
result = cursor.fetchone()
self.close(conn, cursor)
return result
def update_one(self, sql, args):
"""更新数据"""
self.execute(sql, args, isNeed=True)
def insert_all(self, sql, datas):
"""插入多条批量插入"""
conn, cursor = self.open()
try:
cursor.executemany(sql, datas)
conn.commit()
return {'result': True, 'id': int(cursor.lastrowid)}
except Exception as err:
conn.rollback()
print(err)
return {'result': False, 'err': err}
def insert(self,sql,data):
self.execute(sql, data, isNeed=True)
def execute(self, sql, args, isNeed=False):
"""
执行
:param isNeed 是否需要回滚
"""
conn, cursor = self.open()
if isNeed:
try:
cursor.execute(sql, args)
conn.commit()
except:
conn.rollback()
else:
cursor.execute(sql, args)
conn.commit()
self.close(conn, cursor)
import datetime
import sched
import time
import pyodbc
from sqlConfig import MySQLConnectionPool
def time_printer():
now = datetime.datetime.now()
print(now)
# 连接数据库
conn = pyodbc.connect(r'Driver={Microsoft Access Driver (*.mdb, *.accdb)};DBQ=D:\东和.accdb')
# 创建连接池对象
ConnPool = MySQLConnectionPool()
# 创建游标
cursor = conn.cursor()
# 查询数据
cursor.execute('SELECT * FROM 配方列表')
# 清空
sql_update_one = "delete from plan_peifang_list where 1=%s"
ConnPool.update_one(sql_update_one, (1,))
list1 = []
# 逐行读取数据
for row in cursor.fetchall():
print(row[0])
sql1 = "INSERT INTO plan_peifang_list (dingdanhao, kehu, zhilianghao, zhongliang, time, waibuzhilianghao) VALUES (%s, %s, %s, %s, %s, %s)"
ConnPool.insert(sql1, (row[0], row[1], row[2], row[3], row[4], row[5]))
list1.append(row)
loop_monitor()
def loop_monitor():
s = sched.scheduler(time.time,time.sleep)
s.enter(5,1,time_printer,())
s.run()
if __name__ == "__main__":
loop_monitor()
这个定时任务执行一段时间后报错:Traceback (most recent call last):
File "D:\pythonProject.venv\Lib\site-packages\dbutils\pooled_db.py", line 336, in connection
con = self._idle_cache.pop(0)
^^^^^^^^^^^^^^^^^^^^^^^
IndexError: pop from empty list
During handling of the above exception, another exception occurred:
Traceback (most recent call last): 请问是哪里写的不对?连接数用了没关吗?