Python常用模块1-Python的pymysql模块

Bertha 。 2022-11-15 15:54 268阅读 0赞

### 文章目录 ###

*  一.安装
 *  二.pymysql参数介绍
 *  三.pymsql实操
 *   *  3.1 通过pymsql实现增删改查MySQL数据
     *  3.2 通过pymsql实现将csv数据录入到mysql
     *   *  3.2.1 一个insert+update例子
         *  3.2.2 csv空值处理
     *  3.3 通过pymysql将本地mysql数据迁移到远程mysql
 *  参考

# 一.安装 #

Python3.x 如果想连接MySQL需要安装 pymysql 模块。  
pymysql 模块可以通过 pip 安装。

pip install pymysql

# 二.pymysql参数介绍 #

pymysql的参数如下:

#创建数据库连接pymysql.Connect()参数说明
    host(str):      MySQL服务器地址
    port(int):      MySQL服务器端口号
    user(str):      用户名
    passwd(str):    密码
    db(str):        数据库名称
    charset(str):   连接编码，存在中文的时候，连接需要添加charset='utf8'，否则中文显示乱码。
    
    connection对象支持的方法
    cursor()        使用该连接创建并返回游标
    commit()        提交当前事务，不然无法保存新建或者修改的数据
    rollback()      回滚当前事务
    close()         关闭连接
    
    cursor对象支持的方法
    execute(op)     执行SQL，并返回受影响行数
    fetchone()      取得结果集的下一行
    fetchmany(size) 获取结果集的下几行
    fetchall()      获取结果集中的所有行
    rowcount()      返回数据条数或影响行数
    close()         关闭游标对象

# 三.pymsql实操 #

## 3.1 通过pymsql实现增删改查MySQL数据 ##

如需要scott用户下建表及录入数据语句，可参考:  
[scott建表及录入数据sql脚本][scott_sql]

下面是测试通过pymysql来对mysql进行增删改查以及事物。

**代码:**  
db\_setting.py

# 存储数据库信息
    # 创建一个列表嵌套列表,存储数据库信息
    # 依次是 数据库备注、ip、端口、、用户名、密码、登陆数据库名
    
    mysql_msg = [['本地mysql', 'localhost', 3306 , 'root', 'abc123', 'zqs'],
                 ['远程mysql', '10.31.1.123', 3306, 'root', 'abc123', 'test']
                 ]

mysql\_test1.py

import pymysql, datetime
    import db_setting
    
    class mysql_class(object):
        """对mysql的操作创建一个类"""
        def __init__(self, name):
            self.name = name
    
        def login_mysql(self):
            """获取mysql的登陆信息"""
            mysql_msg = db_setting.mysql_msg
            name = self.name
            for i in range(0,len(mysql_msg) ):
                exec_db = mysql_msg[i]
                if name == exec_db[0]:
                    # 根据输入连接数据库
                    return exec_db
                    # conn = pymysql.connect(host=exec_db[1], port=exec_db[2], user=exec_db[3], passwd=exec_db[4],db=exec_db[5])
                else:
                    pass
    
        def insert_mysql(self, datas):
            """往bonus表录入数据"""
    
            exec_db = mysql_class.login_mysql(self)
            conn = pymysql.connect(host=exec_db[1], port=exec_db[2], user=exec_db[3], passwd=exec_db[4], db=exec_db[5])
    
            cursor = conn.cursor()
            sql1 = "insert into bonus(ename, job, sal, comm ) values ('%s', '%s', %.2f, %.2f)"
    
            # 遍历列表,循环录入数据
            for data in datas:
                cursor.execute(sql1 % data)
                print('成功插入', cursor.rowcount, '条数据')
    
            conn.commit()
    
            # 关闭连接
            cursor.close()
            conn.close()
    
        def update_mysql(self, datas):
            """更改bonus表数据"""
    
            exec_db = mysql_class.login_mysql(self)
            conn = pymysql.connect(host=exec_db[1], port=exec_db[2], user=exec_db[3], passwd=exec_db[4], db=exec_db[5])
    
            cursor = conn.cursor()
            sql1 = "update bonus set comm = %.2f where ename = '%s' "
    
            # 遍历列表,循环更改数据
            for data in datas:
                cursor.execute(sql1 % data)
                print('成功修改', cursor.rowcount, '条数据')
    
            conn.commit()
    
            # 关闭连接
            cursor.close()
            conn.close()
    
        def select_mysql(self, datas):
            """查询bonus表数据"""
            exec_db = mysql_class.login_mysql(self)
            conn = pymysql.connect(host=exec_db[1], port=exec_db[2], user=exec_db[3], passwd=exec_db[4], db=exec_db[5])
    
            cursor = conn.cursor()
            sql1 = "select ename,comm from bonus where ename = '%s' "
    
            # 遍历列表,循环查询数据
            for data in datas:
                cursor.execute(sql1 % data)
                for row in cursor.fetchall():
                    print("ename:%s\tcomm:%.2f" % row)
                print('共查找出', cursor.rowcount, '条数据')
    
            conn.commit()
    
            # 关闭连接
            cursor.close()
            conn.close()
    
        def transaction_mysql(self, datas):
            """ 测试mysql的事务 """
    
            exec_db = mysql_class.login_mysql(self)
            conn = pymysql.connect(host=exec_db[1], port=exec_db[2], user=exec_db[3], passwd=exec_db[4], db=exec_db[5])
    
            cursor = conn.cursor()
    
    
            sql1 = "update bonus set comm = %.2f where ename = '%s' "
            sql2 = "update emp set comm = %.2f where ename = '%s' "
    
            try:
                # 遍历列表,循环更改数据
                for data in datas:
                    cursor.execute(sql1 % data)
                    cursor.execute(sql2 % data)
                    print('成功修改', cursor.rowcount, '条数据')
            except Exception as e:
                conn.rollback() # 事务回滚
                print('事务处理失败', e)
            else:
                conn.commit() # 事务提交
                print('事务处理成功', cursor.rowcount)
    
            # 关闭连接
            cursor.close()
            conn.close()
    
        def delete_mysql(self, datas):
            """删除bonus表数据"""
    
            exec_db = mysql_class.login_mysql(self)
            conn = pymysql.connect(host=exec_db[1], port=exec_db[2], user=exec_db[3], passwd=exec_db[4], db=exec_db[5])
    
            cursor = conn.cursor()
            sql1 = "delete from bonus where ename = '%s' "
    
            # 遍历列表,循环删除数据
            for data in datas:
                cursor.execute(sql1 % data)
                print('成功删除', cursor.rowcount, '条数据')
    
            conn.commit()
    
            # 关闭连接
            cursor.close()
            conn.close()
    
    
    
    # 这里才是主程序
    if __name__ == '__main__':
        my_class1 = mysql_class('本地mysql')
    
        # 插入多条数据
        my_class1.insert_mysql([('SMITH', 'ClERK', 800, 100),('ALLEN', 'SALESMAN', 1600, 500)])
    
        # 更改多条数据
        my_class1.update_mysql([(200, 'SMITH'), (600, 'ALLEN')])
    
        # 查询多条数据
        my_class1.select_mysql((('SMITH'),('ALLEN')))
    
        # 事务
        my_class1.transaction_mysql([(300, 'SMITH'), (700, 'ALLEN')])
    
        # 删除多条数据
        my_class1.delete_mysql([('SMITH'),('ALLEN')])

测试记录:

E:\python\learn_python1\venv\Scripts\python.exe E:/python/learn_python1/mysql_test1.py
    成功插入 1 条数据
    成功插入 1 条数据
    成功修改 1 条数据
    成功修改 1 条数据
    ename:SMITH	comm:200.00
    共查找出 1 条数据
    ename:ALLEN	comm:600.00
    共查找出 1 条数据
    成功修改 1 条数据
    成功修改 1 条数据
    事务处理成功 1
    成功删除 1 条数据
    成功删除 1 条数据
    
    Process finished with exit code 0

## 3.2 通过pymsql实现将csv数据录入到mysql ##

生产环境中，我们经常会遇到将csv的数据录入到mysql数据库，类似navicat、mysql workbench等可视化工具提供了excel导入到数据库的功能，但是有时候会报错，本次我们来通过案例讲解将csv的数据录入到mysql。

### 3.2.1 一个insert+update例子 ###

csv文件:

E:\python\file_test>more emp_csv.csv
    "empno","ename","job","mgr","hiredate","sal","comm","deptno"
    "7369","SMITH","CLERK","7902","1980-12-17","800.00","300.00","20"
    "7499","ALLEN","SALESMAN","7698","1981-02-20","1600.00","700.00","30"
    "7521","WARD","SALESMAN","7698","1981-02-22","1250.00","500.00","30"
    "7566","JONES","MANAGER","7839","1981-04-02","2975.00",,"20"
    "7654","MARTIN","SALESMAN","7698","1981-09-28","1250.00","1400.00","30"
    "7698","BLAKE","MANAGER","7839","1981-05-01","2850.00",,"30"
    "7782","CLARK","MANAGER","7839","1981-06-09","2450.00",,"10"
    "7788","SCOTT","ANALYST","7566","1987-06-13","3000.00",,"20"
    "7839","KING","PRESIDENT",,"1981-11-17","5000.00",,"10"
    "7844","TURNER","SALESMAN","7698","1981-09-08","1500.00","0.00","30"
    "7876","ADAMS","CLERK","7788","1987-06-13","1100.00",,"20"
    "7900","JAMES","CLERK","7698","1981-12-03","950.00",,"30"
    "7902","FORD","ANALYST","7566","1981-12-03","3000.00",,"20"
    "7934","MILLER","CLERK","7782","1982-01-23","1300.00",,"10"

用excel打开csv文件截图:  
![image.png][]

创建一个空的emp\_csv表:

create table emp_csv like emp;

**Python代码:**  
db\_setting.py

mysql\_test2.py

import pymysql, datetime
    import csv
    import db_setting
    import math
    from numpy import nan as NaN
    
    class mysql_class(object):
        """对mysql的操作创建一个类"""
        def __init__(self, name):
            self.name = name
    
        def login_mysql(self):
            """获取mysql的登陆信息"""
            mysql_msg = db_setting.mysql_msg
            name = self.name
            for i in range(0,len(mysql_msg) ):
                exec_db = mysql_msg[i]
                if name == exec_db[0]:
                    # 根据输入连接数据库
                    return exec_db
                    # conn = pymysql.connect(host=exec_db[1], port=exec_db[2], user=exec_db[3], passwd=exec_db[4],db=exec_db[5])
                else:
                    pass
    
        def csv_insert_mysql(self):
            """将csv的文件录入到mysql"""
    
            exec_db = mysql_class.login_mysql(self)
            conn = pymysql.connect(host=exec_db[1], port=exec_db[2], user=exec_db[3], passwd=exec_db[4], db=exec_db[5] ,charset='utf8')
    
            cursor = conn.cursor()
    
            # sql1 = "insert into emp_csv(empno, ename, job, mgr, hiredate, sal, comm, deptno) values (%.2f, '%s', '%s', %.2f, '%s', %.2f, %.2f, %.2f)"
    
            # 测试了数值类型还是会存在问题, yyyy/mm/dd 这类时间格式测试了也不会有问题的
            sql1 = "insert into emp_csv(empno, ename, job, mgr, hiredate, sal, comm, deptno) values ('%s', '%s', '%s', '%s', '%s', '%s', '%s', '%s') "
    
            sql2 = "update emp_csv set mgr = null where empno = '%s' "
            sql3 = "update emp_csv set comm = null where empno = '%s' "
    
    
            filename = 'E:/python/file_test/emp_csv.csv'
    
            with open(filename, 'r') as f:
                reader = csv.reader(f)
                # print(type(reader))
    
                for i, row in enumerate(reader):
                    # 剔除第一行的列头
                    if i > 0:
                        # print(row)
                        row3 = tuple(row)
    
                        # 录入数值型空值会有问题,如遇到先改为0 后面再update
                        if row[3] == '':
                            row[3] = '0'
    
                        if row[6] == '':
                            row[6] = '0'
    
                        # 数据库执行的时候,要求是tuple而不是list,进行转换
                        row2 = tuple(row)
    
                        print(row2)
    
                        # 遍历列表,循环录入数据
                        cursor.execute(sql1 % row2)
    
                        # python对数值的空值处理还是存在问题,通过sql进行update吧
                        if row3[3] == '':
                            cursor.execute(sql2 % row3[0])
    
                        if row3[6] == '':
                            cursor.execute(sql3 % row3[0])
    
                conn.commit()
    
                # 关闭连接
                cursor.close()
                conn.close()
    
    
    if __name__ == '__main__':
        my_class1 = mysql_class('本地mysql')
    
        my_class1.csv_insert_mysql()

### 3.2.2 csv空值处理 ###

**对于CSV中空值的处理**  
数值型空值，必须是 %s 而不能使用 ‘%s’  
字符型空值以及时间类型空值，必须是 %s 而不能使用 ‘%s’,但是录入其它非空的字符，又必须用’%s’

用excel打开csv文件截图:  
![image.png][image.png 1]

从上图可以看到 ename列是字符型空值，mgr、comm列是数值型空值，hiredate是日期型空值  
**Python代码:**  
db\_setting.py

mysql\_test3.py

import pymysql, datetime
    import csv
    import db_setting
    import math
    from numpy import nan as NaN
    
    class mysql_class(object):
        """对mysql的操作创建一个类"""
        def __init__(self, name):
            self.name = name
    
        def login_mysql(self):
            """获取mysql的登陆信息"""
            mysql_msg = db_setting.mysql_msg
            name = self.name
            for i in range(0,len(mysql_msg) ):
                exec_db = mysql_msg[i]
                if name == exec_db[0]:
                    # 根据输入连接数据库
                    return exec_db
                    # conn = pymysql.connect(host=exec_db[1], port=exec_db[2], user=exec_db[3], passwd=exec_db[4],db=exec_db[5])
                else:
                    pass
    
        def csv_insert_mysql(self):
            """将csv的文件录入到mysql"""
    
            exec_db = mysql_class.login_mysql(self)
            conn = pymysql.connect(host=exec_db[1], port=exec_db[2], user=exec_db[3], passwd=exec_db[4], db=exec_db[5] ,charset='utf8')
    
            cursor = conn.cursor()
    
            # 测试了数值类型还是会存在问题, yyyy/mm/dd 这类时间格式测试了也不会有问题的
            # 数值型空值，必须是 %s 而不能使用 '%s'
            # 字符型空值以及时间类型空值，必须是 %s 而不能使用 '%s',但是录入其它非空的字符，又必须用'%s'
            sql = ""
            sql1 = "insert into emp_csv(empno, ename, job, mgr, hiredate, sal, comm, deptno) values ('%s', %s, '%s', %s, '%s', '%s', %s, '%s') "
            sql2 = "insert into emp_csv(empno, ename, job, mgr, hiredate, sal, comm, deptno) values ('%s', '%s', '%s', %s, '%s', '%s', %s, '%s') "
    
            sql3 = "insert into emp_csv(empno, ename, job, mgr, hiredate, sal, comm, deptno) values ('%s', '%s', '%s', %s, %s, '%s', %s, '%s') "
    
    
            filename = 'E:/python/file_test/emp_csv_2.csv'
    
            with open(filename, 'r') as f:
                reader = csv.reader(f)
                # print(type(reader))
    
                for i, row in enumerate(reader):
                    # 剔除第一行的列头
                    if i > 0:
                        empno = row[0]
    
                        if  row[1]:
                            ename = row[1]
                        else:
                            ename = 'NULL'
    
                        job = row[2]
    
                        if  row[3]:
                            mgr = row[3]
                        else:
                            mgr = 'NULL'
    
                        if  row[4]:
                            hiredate = row[4]
                        else:
                            hiredate = 'NULL'
    
                        sal = row[5]
    
                        if row[6]:
                            comm = row[6]
                        else:
                            comm =  'NULL'
    
                        deptno = row[7]
    
                        print(empno, ename, job, mgr , hiredate, sal, comm , deptno)
    
                        # 遍历列表,循环录入数据
                        if not row[1]:
                            sql = sql1
    
                        if not row[4]:
                            sql = sql3
    
                        # 如果存在两列同时为null的情况,还得继续增加判断条件
    
                        if row[1]:
                            if row[4]:
                                sql = sql2
    
                        cursor.execute(sql % (empno, ename, job, mgr, hiredate, sal, comm, deptno))
    
    
    
                conn.commit()
    
                # 关闭连接
                cursor.close()
                conn.close()
    
    
    if __name__ == '__main__':
        my_class1 = mysql_class('本地mysql')
    
        my_class1.csv_insert_mysql()

## 3.3 通过pymysql将本地mysql数据迁移到远程mysql ##

其实原理很简单，先在源端mysql将数据存储在列表中，然后再到目标mysql进行insert操作，需要注意的是空值的处理

**代码:**  
db\_setting.py

mysql\_test4.py

import pymysql, datetime
    import csv
    import db_setting
    import math
    from numpy import nan as NaN
    
    
    
    def login_mysql(mysql_name):
        """获取mysql的登陆信息"""
        mysql_msg = db_setting.mysql_msg
        name = mysql_name
        for i in range(0,len(mysql_msg) ):
            exec_db = mysql_msg[i]
            if name == exec_db[0]:
                # 根据输入连接数据库
                return exec_db
                # conn = pymysql.connect(host=exec_db[1], port=exec_db[2], user=exec_db[3], passwd=exec_db[4],db=exec_db[5])
            else:
                pass
    
    def t_data():
        """将本地数据库数据同步到远程数据库"""
        exec_db = login_mysql('本地mysql')
    
        conn = pymysql.connect(host=exec_db[1], port=exec_db[2], user=exec_db[3], passwd=exec_db[4], db=exec_db[5],charset='utf8')
    
        cursor = conn.cursor()
    
        mysql_result1 = [];
    
        sql1 = "select empno, ename, job, mgr, hiredate, sal, comm, deptno from emp"
    
        cursor.execute(sql1)
        for row in cursor.fetchall():
            mysql_result1.append(row)
    
        # print(mysql_result1)
    
        # 关闭连接
        cursor.close()
        conn.close()
    
    
        # 开始连接目标mysql服务器
        exec_db = login_mysql('远程mysql')
    
        conn = pymysql.connect(host=exec_db[1], port=exec_db[2], user=exec_db[3], passwd=exec_db[4], db=exec_db[5],charset='utf8')
    
        cursor = conn.cursor()
    
        sql2 = "insert into emp(empno, ename, job, mgr, hiredate, sal, comm, deptno) values ('%s', '%s', '%s', %s, '%s', '%s', %s, '%s')"
    
        for row in mysql_result1:
            # 数值型的null值，中转后变为None，执行insert的时候，默认是会是0 或 0.00 ,这个地方必须中转一下
            if row[3]:
                mgr = row[3]
            else:
                mgr = 'NULL'
    
            if row[6]:
                comm = row[6]
            else:
                comm = 'NULL'
    
            cursor.execute(sql2 % (row[0],row[1],row[2],mgr,row[4],row[5],comm,row[7]))
            # print(row)
    
        conn.commit()
    
        # 关闭连接
        cursor.close()
        conn.close()
    
    if __name__ == '__main__':
        t_data()

# 参考 #

1.https://www.cnblogs.com/yuhou/p/10893056.html  
2.https://www.cnblogs.com/liangshian/p/11272155.html

[scott_sql]: https://www.jianshu.com/p/532fe68924cb
[image.png]: /images/20221022/275afa1ec1a444969b7bf0121934551c.png
[image.png 1]: /images/20221022/6d2a21bba5b84dfeb3c08b9433d8ed7a.png