python代码执行SQL文件(逐句执行)

古城微笑少年丶 2024-04-18 15:30 45阅读 0赞

# 一、简介 #

关于Python如何连接数据库并执行SQL语句，几乎所有的Python教程都会讲，教程里基本只介绍了执行单条SQL语句的方法，但是实际生产过程中可不只是执行一两条语句，动辄几十条甚至上百条的SQL语句，一个.SQL的SQL文件正常都会包含多条SQL语句，如果把这些SQL语句直接贴进Python代码文件，势必会影响代码的可读性和美感，代码量也冗余，那么Python能不能像oracle的PL/SQL客户端或者navicat一样读取SQL文件并且执行会是一件非常有趣的事情。

经过不断百度和尝试，一段优雅的Python代码终于敲出来，作为开源的忠实粉丝，技术始于分享，现将Python执行SQL文件的代码分享在CSDN，代码实例是MySQL数据库。

# 二、原理实现 #

我们知道，mysql的语句执行时，默认是以分号结尾的。因此，只要将整个sql文件读取，以分号切割，就可以得到sql语句。

sql文件，内容如下：

USE test;
    /*迁移用户信息*/
    INSERT INTO `users` (
        `UID`,
        `USERNAME`,
        `PASSWORD`,
        `ROLE`,
        `CREATE_TIME`
    ) SELECT
        `id`,
        `username`,
        `password`,
        `level`,
        `create_time`
    FROM
        tbl_users;

使用python切割以分号切割，打印的sql还是很多行。因此需要将换行符，替换为空格。末尾增加分号结尾

with open('test.sql',encoding='utf-8',mode='r') as f:
        # 读取整个sql文件，以分号切割。[:-1]删除最后一个元素，也就是空字符串
        sql_list = f.read().split(';')[:-1]
        for x in sql_list:
            # 判断包含空行的
            if '\n' in x:
                # 替换空行为1个空格
                x = x.replace('\n', ' ')
    
            # sql语句添加分号结尾
            sql_item = x+';'
            print(sql_item)

执行输出：

USE test;
     /*迁移用户信息*/ INSERT INTO `users` (     `UID`,     `USERNAME`,     `PASSWORD`,     `ROLE`,     `CREATE_TIME` ) SELECT     `id`,     `username`,     `password`,     `level`,     `create_time` FROM     tbl_users;

可以发现，会有很多空行。可以再加一个判断，进行替换

with open('test.sql',encoding='utf-8',mode='r') as f:
        # 读取整个sql文件，以分号切割。[:-1]删除最后一个元素，也就是空字符串
        sql_list = f.read().split(';')[:-1]
        for x in sql_list:
            # 判断包含空行的
            if '\n' in x:
                # 替换空行为1个空格
                x = x.replace('\n', ' ')
    
            # 判断多个空格时
            if '    ' in x:
                # 替换为空
                x = x.replace('    ', '')
    
            # sql语句添加分号结尾
            sql_item = x+';'
            print(sql_item)

执行输出：

USE test;
     /*迁移用户信息*/ INSERT INTO `users` ( `UID`, `USERNAME`, `PASSWORD`, `ROLE`, `CREATE_TIME` ) SELECT `id`, `username`, `password`, `level`, `create_time` FROM tbl_users;

# 三、正式代码 #

上面的步骤已经得到完整的sql语句了，那么就可以使用pymysql模块来执行每一句sql了。

## 环境说明 ##

原始表结构为：

CREATE TABLE `tbl_users` (
      `id` int(11) NOT NULL AUTO_INCREMENT,
      `username` varchar(20) NOT NULL DEFAULT '' COMMENT '用户名',
      `password` char(32) NOT NULL DEFAULT '' COMMENT '密码',
      `level` tinyint(10) NOT NULL DEFAULT '2' COMMENT '用户等级',
      `create_time` datetime DEFAULT NULL COMMENT '创建时间',
      PRIMARY KEY (`id`)
    ) ENGINE=MyISAM DEFAULT CHARSET=utf8 COMMENT='用户表';

目标表结构为：

CREATE TABLE `users` (
      `UID` bigint(20) NOT NULL AUTO_INCREMENT,
      `USERNAME` varchar(20) NOT NULL DEFAULT '' COMMENT '用户名',
      `PASSWORD` char(32) NOT NULL DEFAULT '' COMMENT '密码',
      `ROLE` varchar(20) NOT NULL DEFAULT '' COMMENT '角色',
      `CREATE_TIME` datetime DEFAULT NULL COMMENT '创建时间',
      PRIMARY KEY (`UID`)
    ) ENGINE=MyISAM DEFAULT CHARSET=utf8 COMMENT='用户表';

需要将原始表的数据，迁移到目标表中。

mysql服务器ip为 192.168.10.104，用户名和密码都是root，使用test数据库。

## 完整代码 ##

完整代码如下：

#!/usr/bin/env python3
    # coding: utf-8
    import pymysql
    
    try:
        db = pymysql.connect("192.168.10.104", "root", "root", "test")
        c = db.cursor()
        with open('test.sql',encoding='utf-8',mode='r') as f:
            # 读取整个sql文件，以分号切割。[:-1]删除最后一个元素，也就是空字符串
            sql_list = f.read().split(';')[:-1]
            for x in sql_list:
                # 判断包含空行的
                if '\n' in x:
                    # 替换空行为1个空格
                    x = x.replace('\n', ' ')
    
                # 判断多个空格时
                if '    ' in x:
                    # 替换为空
                    x = x.replace('    ', '')
    
                # sql语句添加分号结尾
                sql_item = x+';'
                # print(sql_item)
                c.execute(sql_item)
                print("执行成功sql: %s"%sql_item)
    except Exception as e:
        print(e)
        print('执行失败sql: %s'%sql_item)
    finally:
        # 关闭mysql连接
        c.close()
        db.commit()
        db.close()

执行输出：

执行成功sql: USE test;
    执行成功sql:  /*迁移用户信息*/ INSERT INTO `users` ( `UID`, `USERNAME`, `PASSWORD`, `ROLE`, `CREATE_TIME` ) SELECT `id`, `username`, `password`, `level`, `create_time` FROM tbl_users;

本文参考链接：

[https://blog.csdn.net/jsyzliuyu/article/details/79632995][https_blog.csdn.net_jsyzliuyu_article_details_79632995]

[https_blog.csdn.net_jsyzliuyu_article_details_79632995]: https://blog.csdn.net/jsyzliuyu/article/details/79632995