Background
公司有业务需求需要在MySQL中批量创建表,于是用python封装了个工具类进行操作,记录下来,不定以后用的到。表名和表的字段如下图所示。
python脚本如下
from pymysql import connect
# 一、创建传感器测点表
class MySQLUtil(object):
# 初始化方法
def __init__(self, host, database='default', user='root', password='123456', charset='utf8', port=3306):
self.host = host
self.database = database
self.user = user
self.password = password
self.charset = charset
self.port = port
self.conn = None
self.cur = None
# 初始化完后连接到数据库并创建好游标
def open(self):
# 连接数据库
self.conn = connect(host=self.host, user=self.user, password=self.password, database=self.database,
charset=self.charset, port=self.port)
# 创建游标
self.cur = self.conn.cursor()
# 关闭数据库
def close(self):
self.cur.close()
self.conn.close()
# 数据库操作 增删改
def sql_execute(self, sql):
# 打开数据库
self.open()
# 对sql语句处理(即开始操作数据库了),有成功有失败
try:
# 执行sql命令
self.cur.execute(sql)
# 提交到数据库执行
self.conn.commit()
except Exception as e:
self.conn.rollback()
print('failed', e)
# 关闭数据库
self.close()
# 判断表是否存在
def table_exists(self, table_name):
self.open()
sql = "SHOW TABLES LIKE '%s';" % table_name
self.cur.execute(sql)
tables = self.cur.fetchone()
self.close()
if tables is not None:
return 1
else:
return 0
# 查询
def get_quota_count(self, target_id):
self.open()
sql_get = "SELECT COUNT(t2.code) FROM t_target t1 LEFT JOIN t_settings_data_quota t2 " \
"ON t1.type = t2.target_type WHERE t1.id = %s" % target_id
self.cur.execute(sql_get)
# 查询结果用result绑定
result = self.cur.fetchone()
self.close()
# 将查询到的结果返回回去
return result[0]
# 批量创建表
def create_tables(self, tbl_start_num, tbl_end_num):
# 执行 sql 语句
for target_id in range(tbl_start_num, tbl_end_num + 1):
# 获取 target_id 所对应类型的 quota 个数,以此决定所创建表字段的个数
count = self.get_quota_count(target_id)
# 所创建表字段的个数
fields = ""
for i in range(count):
i = i + 1
fields = fields + "value%s decimal(10, 5) NULL DEFAULT NULL COMMENT '数值%s'," % (i, i)
# 所创建表的表名
tbl = "z_data_" + str(target_id)
sql_create = "CREATE TABLE %s (`id` int(11) NOT NULL AUTO_INCREMENT COMMENT 'ID'," \
"`target_id` int(11) NULL DEFAULT NULL COMMENT '测点ID'," \
"`time_o` timestamp(0) NULL DEFAULT CURRENT_TIMESTAMP COMMENT '数据采集时间'," \
"`time_w` timestamp(0) NULL DEFAULT CURRENT_TIMESTAMP COMMENT '数据入库时间'," % tbl \
+ fields + \
"`is_test` int(11) NULL DEFAULT 0 COMMENT '0:生产数据,1:测试数据'," \
"PRIMARY KEY (`id`) USING BTREE, INDEX `time_o`(`time_o`) USING BTREE);"
# 执行 sql
try:
self.sql_execute(sql_create)
print(tbl + ":创建成功!")
except Exception as e:
print(tbl + ":创建失败!")
# 二、删除传感器测点表
def del_tables(self, tbl_start_num, tbl_end_num):
# 执行 sql 语句
for i in range(tbl_start_num, tbl_end_num + 1):
tbl = "z_data_" + str(i)
sql_delete = "DROP TABLE %s;" % tbl
# 执行 sql
try:
self.sql_execute(sql_delete)
print(tbl + ":删除成功!")
except Exception as e:
print(tbl + ":创建失败!")
if __name__ == '__main__':
# 测试环境
host_ip = '192.168.10.110'
# 生产环境
# host_ip = '14.16.0.211'
# 测试和生产 MySQL 连接基础信息一样
db = 'db_test'
name = 'root'
pwd = '123456'
# 获取数据库连接
mysql_util = MySQLUtil(host_ip, db, name, pwd)
# 【1、创建表】 ---创建和删除需传入表的 target_id (左右边界包含)
mysql_util.create_tables(338, 427)
# 【2、删除表】 ---创建和删除需传入表的 target_id (左右边界包含)
# mysql_util.del_tables(400, 402)
# 【3、判断表是否存在】
# print(mysql_util.table_exists("z_data_4000"))