#!/usr/bin/env python # -*- encoding: utf-8 -*- ''' @Date:2022/08/26 08:34:30 ''' # import sys # sys.path.append('.') # import pymysql # import pymysql # conn = pymysql.connect(host='127.0.0.1', port=3306, # user='root', passwd='123456', db='local_rms_db') # cur = conn.cursor(cursor=pymysql.cursors.DictCursor) # sql = "select * from rms_user" # cur.execute(sql) # data = cur.fetchall() # print(data) # import os # from flask import Flask, jsonify # app = Flask(__name__) # def get_file_base(): # import base64 # with open(os.getcwd() + "/config/tem_00001.xlsx", "rb") as f: # file_str = f.read() # file_str = base64.b64encode(file_str).decode() # return file_str # @app.route("/get_file", methods=["GET"]) # def main_file(): # return jsonify(get_file_base()) # if __name__ == '__main__': # app.run("127.0.0.1", 8001) db_name = "user" data_list = [ {"variety_id": "f8442a6e-9950-11ec-a502-e2052a1045e4", "remark1": "", "remark2": "", "remark3": "\u6613\u5236\u6bd2", "remark4": "", "name": "\u7532\u82ef", "english_name": "", "purity": "AR\uff08\u6caa\u8bd5\uff09\uff0c\u2265 99.5%", "cas_number": "108-88-3", "remark5": "", "speci": "500", "speci_unit": "ml", "export_count": "1", "remark6": "\u74f6", "production_date": "2022-03-01", "shelf_life": 10529, "price": "0", "is_supervise": 0, "remain": "683", "manufacturer": "\u56fd\u836f", "remark8": "", "remark9": "\u70f7\u57fa\u6c5e", "remark10": ""}, {"variety_id": "f8442a6f-9950-11ec-9253-e2052a1045e4", "remark1": "", "remark2": "", "remark3": "\u5371\u5316\u54c1", "remark4": "", "name": "\u82ef\u915a", "english_name": "", "purity": "AR\uff08\u6caa\u8bd5\uff09\uff0c\u2265 99.0%", "cas_number": "108-95-2", "remark5": "", "speci": "500", "speci_unit": "ml", "export_count": "1", "remark6": "\u74f6", "production_date": "2022-03-01", "shelf_life": 10529, "price": "0", "is_supervise": 0, "remain": "848", "manufacturer": "\u56fd\u836f", "remark8": "", "remark9": "\u6325\u53d1\u915a", "remark10": ""}, {"variety_id": "f8442a70-9950-11ec-a157-e2052a1045e4", "remark1": "", "remark2": "", "remark3": "\u5371\u5316\u54c1", "remark4": "", "name": "\u82ef", "english_name": "", "purity": "AR\uff08\u6caa\u8bd5\uff09\uff0c\u2265 99.5%", "cas_number": "71-43-2", "remark5": "", "speci": "500", "speci_unit": "ml", "export_count": "1", "remark6": "\u74f6", "production_date": "2022-03-01", "shelf_life": 10529, "price": "0", "is_supervise": 0, "remain": "640", "manufacturer": "\u56fd\u836f", "remark8": "", "remark9": "\u6cb9", "remark10": ""}, {"variety_id": "f8442a71-9950-11ec-9894-e2052a1045e4", "remark1": "", "remark2": "", "remark3": "\u5371\u5316\u54c1", "remark4": "", "name": "\u4e59\u9178\uff0836%\uff09", "english_name": "", "purity": "AR", "cas_number": "64-19-7", "remark5": "", "speci": "500", "speci_unit": "ml", "export_count": "1", "remark6": "\u74f6", "production_date": "2022-03-01", "shelf_life": 10529, "price": "0", "is_supervise": 0, "remain": "734", "manufacturer": "\u56fd\u836f", "remark8": "", "remark9": "\u70f7\u57fa\u6c5e", "remark10": ""}, {"variety_id": "f84451a6-9950-11ec-8925-e2052a1045e4", "remark1": "", "remark2": "", "remark3": "\u5371\u5316\u54c1", "remark4": "", "name": "\u4e59\u9178\u9150", "english_name": "", "purity": "AR", "cas_number": "106-24-7", "remark5": "", "speci": "500", "speci_unit": "ml", "export_count": "1", "remark6": "\u74f6", "production_date": "2022-03-01", "shelf_life": 10529, "price": "0", "is_supervise": 0, "remain": "749", "manufacturer": "\u56fd\u836f", "remark8": "", "remark9": "\u70f7\u57fa\u6c5e", "remark10": ""}, {"variety_id": "f84451a7-9950-11ec-9b4e-e2052a1045e4", "remark1": "", "remark2": "", "remark3": "\u5371\u5316\u54c1", "remark4": "", "name": "\u56db\u6c2f\u4e59\u70ef", "english_name": "", "purity": "\u73af\u4fdd\u7ea7", "cas_number": "127-18-4", "remark5": "", "speci": "500", "speci_unit": "ml", "export_count": "28", "remark6": "\u74f6", "production_date": "2022-03-01", "shelf_life": 10529, "price": "0", "is_supervise": 0, "remain": "1207", "manufacturer": "\u5929\u6d25\u50b2\u7136", "remark8": "", "remark9": "\u6cb9", "remark10": ""}, {"variety_id": "f84451a8-9950-11ec-96ca-e2052a1045e4", "remark1": "", "remark2": "", "remark3": "\u5371\u5316\u54c1", "remark4": "", "name": "\u56db\u6c2f\u5316\u78b3", "english_name": "", "purity": "\u7ea2\u5916\u6d4b\u6cb9\u4eea\u4e13\u7528", "cas_number": "56-23-5", "remark5": "", "speci": "500", "speci_unit": "ml", "export_count": "31", "remark6": "\u74f6", "production_date": "2022-03-01", "shelf_life": 10529, "price": "0", "is_supervise": 0, "remain": "1126", "manufacturer": "\u5b89\u8c31", "remark8": "", "remark9": "\u77ff\u7269\u6cb9", "remark10": ""}] # sql_header = f"insert into {}" key_tup = [] value_tup = [] if data_list: base_da = data_list[0] for k,v in base_da.items(): key_tup.append(k) value_tup.append("%s") insert_sql = f"insert into {db_name}({','.join(key_tup)}) values({','.join(value_tup)})" print(insert_sql) "".join import pymysql from dbutils.pooled_db import PooledDB # from common.db.mysql_config import MysqlConfig """ pymysql封装总结 https://blog.csdn.net/zhj_1121/article/details/121070412 python操作mysql之只看这篇就够了 https://www.jianshu.com/p/4e72faebd27f 关于PooledDB使用autocommit的方法 https://blog.51cto.com/abyss/1736844 """ class MysqlPool: """ MySQL 数据库连接池类 配置变量 """ ''' :param reset: how connections should be reset when returned to the pool (False or None to rollback transcations started with begin(), True to always issue a rollback for safety's sake) :param setsession: optional list of SQL commands that may serve to prepare the session, e.g. ["set datestyle to ...", "set time zone ..."] ''' ''' https://blog.51cto.com/abyss/1736844 其中的 setsession=['SET AUTOCOMMIT = 1'] 就是用来设置线程池是否打开自动更新的配置，0为False，1为True ''' # 初始化数据库连接池变量 __pool = None # 创建连接池的最大数量 __MAX_CONNECTIONS = 20 # 连接池中空闲连接的初始数量 __MIN_CACHED = 5 # 连接池中空闲连接的最大数量 __MAX_CACHED = 5 # 共享连接的最大数量 __MAX_SHARED = 0 # 超过最大连接数量时候的表现，为True等待连接数量下降，为false直接报错处理 __BLOCK = True # 单个连接的最大重复使用次数 __MAX_USAGE = 1 # 当返回到池时，连接应该如何重置 # (False或None回滚以begin()开始的事务，为了安全起见，总是发出回滚) __RESET = True # 设置自动提交 __SET_SESSION = [] # 不能是 UTF-8 __CHARSET = 'UTF8' def __init__(self, host, port, user, password, database): """ :param host: 数据库主机地址 :param port: 端口号 :param user: 用户名 :param password: 密码 :param database: 数据库名 """ if not self.__pool: # self代表当前类的实例，即为 MysqlPool（） 带小括号，执行后的数据。 # __class__，魔法函数，代表从当前类的实例中，获取当前类，即为 MysqlPool 不带小括号的类。 # __pool，这个代表的事类的变量，即为在类下面创建的初始化连接池，__pool self.__class__.__pool = PooledDB( creator=pymysql, host=host, port=port, user=user, password=password, database=database, maxconnections=self.__MAX_CONNECTIONS, mincached=self.__MIN_CACHED, maxcached=self.__MAX_CACHED, maxshared=self.__MAX_SHARED, blocking=self.__BLOCK, maxusage=self.__MAX_USAGE, setsession=self.__SET_SESSION, charset=self.__CHARSET ) def get_connect(self): return self.__pool.connection() class MysqlCursor: """ 从数据库配置环境，取出数据库配置参数 这里的参数，可以不从外部导入，直接手动写入也可以。 """ def __init__(self, host=host, port=port, user=user, password=password, database=database) -> None: """ :param host: 数据库主机地址 :param port: 端口号 :param user: 用户名 :param password: 密码 :param database: 数据库名 """ self.__host = host self.__port = port self.__user = user self.__password = password self.__database = database # 初始化数据库连接池 self.connects_pool = MysqlPool( host=self.__host, port=self.__port, user=self.__user, password=self.__password, database=self.__database ) def __enter__(self): """ # with 上下文管理，魔法函数，进入with时调用 :return: 当前类 """ # 从数据库链接池，获取一个数据库链接 connect = self.connects_pool.get_connect() # 从获取的数据库链接，获取一个光标 cursor = connect.cursor(pymysql.cursors.DictCursor) ''' # https://blog.51cto.com/abyss/1736844 # 如果使用连接池 则不能在取出后设置 而应该在创建线程池时设置 # connect.autocommit = False ''' # 将数据库链接，赋值给当前类，方便__exit__函数调用 self._connect = connect # 将数据库光标，赋值给当前类，方便__exit__函数调用 self._cursor = cursor # __enter__函数，必须返回当前类 return self def __exit__(self, *exc_info): """ # with 上下文管理，魔法函数，退出with时调用 :param exc_info: 异常信息，元祖 :return: None """ # 退出with上下文时，使用当前类链接，提交数据库语句 self._connect.commit() # 关闭光标 self._cursor.close() # 关闭链接 self._connect.close() @property def cursor(self): """ 数据库连接池，取出链接，取出光标，转换为光标属性 :return: 数据库连接池的光标 """ return self._cursor if __name__ == "__main__": with MysqlCursor() as db: # 获取数据库的方法 sql = 'select count(id) as total from people' db.cursor.execute("select count(id) as total from people") data = db.cursor.fetchone() print('--------统计数据条数', data)