安装依赖
pip3 install pyodbc -i https://mirrors.aliyun.com/pypi/simple/ requests
如果时linux环境可能需要执行以下操作
yum -y install wget
yum remove -y unixODBC
curl https://packages.microsoft.com/config/rhel/7/prod.repo > /etc/yum.repos.d/mssql-release.repo
ACCEPT_EULA=Y yum install -y msodbcsql17db_pool.py
# -*- coding: utf-8 -*-# ====> sqlserver数据库连接池 <==== #import traceback
from datetime import datetime
from queue import Queueimport pyodbcdef info(msg):print(f'{datetime.now().strftime("%Y-%m-%d %H:%M:%S")}|INFO|{msg}')def error(msg):print(f'{datetime.now().strftime("%Y-%m-%d %H:%M:%S")}|ERROR|{msg}')class ConnectionPool:def __init__(self, server, port, user, password, database, max_connections):self.server = serverself.port = portself.user = userself.password = passwordself.database = databaseself.max_connections = max_connectionsself.connections = Queue(maxsize=max_connections)driver = '{' + pyodbc.drivers()[0] + "}"conn_param = f'DRIVER={driver};SERVER={server},{port};DATABASE={database};UID={user};PWD={password}'# info(f'连接sqlserver参数: {conn_param}')for _ in range(max_connections):conn = pyodbc.connect(conn_param)self.connections.put(conn)info(f'初始化数据库连接池成功, 连接池数量{max_connections}')def get_connection(self):return self.connections.get()def release_connection(self, conn):self.connections.put(conn)# 查询def query(self, sql):conn = self.get_connection()result = []try:cursor = conn.cursor()cursor.execute(sql)result = cursor.fetchall()except Exception as e:error(f'执行查询sql报错, msg:{e} sql:{sql}')traceback.print_exc()self.release_connection(conn)return result# 执行sqldef execute(self, sql):result = Trueconn = self.get_connection()try:cursor = conn.cursor()cursor.execute(sql)conn.commit()except Exception as e:error(f'执行查询sql报错, msg:{e} sql:{sql}')traceback.print_exc()result = Falseself.release_connection(conn)return result# 根据sql批量查询,sql示例: INSERT INTO t1("create_time", "modify_time", "name", "age") VALUES (?,?,?,?)def insert_for_sql(self, sql, data):conn = self.get_connection()result = []cursor = conn.cursor()cursor.executemany(sql, data)conn.commit()self.release_connection(conn)return result# 根据表明和字段列表批量插入def insert_for_obj(self, table, fields, data):conn = self.get_connection()result = []cursor = conn.cursor()cursor.executemany(self.get_insert_sql(table, fields), data)conn.commit()self.release_connection(conn)return result# 根据表名和字段列表生成sqldef get_insert_sql(self, table, fields):f_count = len([item for item in fields if item.strip()])quoted_arr = ['"' + str(item) + '"' for item in fields]sql = f'''INSERT INTO {table}({', '.join(quoted_arr)}) VALUES ({",".join(["?"] * f_count)})'''return sqldef release(self):count = 0while not self.connections.empty():count += 1self.connections.get().close()info(f'数据库连接释放成功，数量{count}')database = 'test_db'
info(f'初始化数据库连接池, 数据库名:{database}')
pool = ConnectionPool('127.0.0.1',1433,'root','123456',database,1)
测试
from datetime import datetimefrom db_pool import pooldef create_table():sql = '''
CREATE TABLE test1(id INT IDENTITY(1,1) PRIMARY KEY, -- 自增主键idcreate_time DATETIME, -- 创建时间modify_time DATETIME, -- 修改时间name VARCHAR(50), -- 名称age INT -- 年龄
);
'''pool.execute(sql)def insert():now = get_now()ds = [(now, now, '神里绫华', 14), (now, now, '申鹤', 15)]fields = ['create_time', 'modify_time', 'name', 'age']pool.insert_for_obj('test1', fields, ds)def query():print('--------------查询--------------')rs = pool.query('select * from test1')for r in rs:print(r)def drop_table():pool.execute('drop table test1')def get_now():# 获取当前时间current_time = datetime.now()# 将当前时间格式化为带毫秒的时间格式formatted_time = current_time.strftime("%Y-%m-%d %H:%M:%S")return formatted_timeif __name__ == '__main__':# create_table()insert()query()#drop_table()pool.release()
注意事项
数据库varchar(5000), db.Column(db.String(1999) 大于1999插入会报精度异常class MyTest(db.Model):id = db.Column(db.Integer, primary_key=True)name = db.Column(db.String(5000)
