使用Python提高工作效率之MySQL

发表于 | 更新于 | 分类于 | 阅读次数：

日常开发测试中经常需要在数据库中造一些测试数据，为了能够快速完成任务，通常我都喜欢用Python来写。目前使用比较多的是MySQL数据库，Python有对应的很多第三方库可以用，但我还是坚持使用MySQL官方提供的支持库mysql-connector-python，因为这个库提供了MySQL的所有操作API，而且本身语法已经足够简单，对于写日常脚本来说已经足够。

准备工作

  • Python3.x环境
  • MySQL 8.0数据库
  • MySQL官方提供的Python驱动

不同的操作系统可选的安装方式都差不多，具体内容可以参照MySQL官方文档
https://dev.mysql.com/doc/connector-python/en/connector-python-installation-binary.html

不过也不一定非要把驱动安装到系统上，比如下载驱动库就可以直接用Python去操作了，可以直接在当前的Python工程Env环境下安装

1
pip install mysql-connector-python

快速使用

连接到数据库

1.引入驱动库

1
import mysql.connector

2.打开/关闭连接

1
2
3
4
5
cnx = mysql.connector.connect(user='root', password='123456',
                              host='127.0.0.1',
                              port='3306',
                              database='py_example')
cnx.close()

mysql.connector.connect()方法提供了很多参数，有些参数是可选的，比如端口号如果不传默认就是3306
3.异常的捕获

1
2
3
4
5
6
7
8
9
10
import mysql.connector

try:
    cnx = mysql.connector.connect(user='root', password='123456',
                                  host='127.0.0.1',
                                  port='3307',
                                  database='py_example')
    cnx.close()
except mysql.connector.Error as err:
    print(err)

所连接的MySQL端口号为3306，在代码中故意将端口号设置为3307,使其无法正常连接到数据库，抛出异常后将被捕获并打印：

1
2003 (HY000): Can't connect to MySQL server on '127.0.0.1' (61)

如果要根据不同的错误码进行自定义输出，可以在头部引入from mysql.connector import errorcode,通过对比捕获的err.errno和errorcode的各种预定义错误值进行比对,更多连接方式和详细的异常捕获输出，可以参照官方文档：
https://dev.mysql.com/doc/connector-python/en/connector-python-example-connecting.html

灵活的连接参数

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
import mysql.connector

config = {
    "user": "root",
    "password": "123456",
    "host": "127.0.0.1",
    "database": "py_example"
}


def connect(conf):
    try:
        cnx = mysql.connector.connect(**conf)
    except mysql.connector.Error as err:
        print(err)
    else:
        return cnx

mysql.connector.connect()方法支持传入一个json对象，这样就可以很方便的把数据库连接定义到外部或者外部文件。
连接数据库是为了得到连接对象，所以定义一个connect()方法，传入参数为数据库配置，该方法返回一个数据库连接对象cnx。

数据库的基本操作

通过mysql.connector.connect()返回一个connect连接对象，通过该连接对象的cursor()方法可以得到操作数据库的游标。

1
2
3
if __name__ == '__main__':
    conn = connect(config)
    cursor = conn.cursor()

通过cursor.execute()方法就可以执行SQL语句了。

创建数据库

1
cursor.execute("CREATE DATABASE {} DEFAULT CHARACTER SET 'utf8mb4'".format('db_name'))

创建表

1
2
3
4
5
6
7
8
9
10
table_sql = """
CREATE TABLE `table_name` (
  `t_id` INT(11) NOT NULL AUTO_INCREMENT,
  `name` VARCHAR(255) NOT NULL ,
  `gender` ENUM('M','F') NOT NULL ,
  `birth_date` DATE NOT NULL,
  PRIMARY KEY (`t_id`)
) ENGINE = INNODB
"""
cursor.execute(table_sql)

插入数据

1
2
3
4
5
6
insert_sql = """
INSERT INTO `table_name` (`name`,`gender`,`birth_date`) VALUES ('蓝士钦','M','1995-01-29')
"""
cursor.execute(insert_sql)
# conn默认不会自动提交事务，所以这里一定要手动提交事务，否则记录不会插入到MySQL中
conn.commit()

除了使用SQL拼接方式，还可以使用模块符代替，然后通过传入一个Python元组的方式

1
2
3
4
5
6
7
insert_sql_template = """
INSERT INTO `table_name` (`name`,`gender`,`birth_date`) VALUES (%s,%s,%s)
"""
insert_sql_date = ('蓝士钦', 'M', date(1995, 1, 29))
cursor.execute(insert_sql_template,insert_sql_date)
# 提交事务
conn.commit()

除了使用元组，还可以使用Json格式的对象作为变量填充参数

1
2
3
4
5
6
7
8
9
10
11
12
13
sql = """
INSERT INTO `table_name` (`name`,`gender`,`birth_date`) VALUES (%(name)s,%(gender)s,%(birth_date)s)
"""

data = {
    "name": "测试姓名",
    "gender": "M",
    "birth_date": date(1995, 1, 29),
}

cursor.execute(sql,data)
# 提交事务
conn.commit()

可以看到代码中每次使用cursor.execute()方法执行sql语句之后，都需要掉用连接对象conn.commit()方法，
conn默认不会自动提交事务，所以这里一定要手动提交事务，否则记录不会插入到MySQL中。
不过也可以根据实际使用场景，设置每次execute都自动提交事务，只需要获取到连接对象后设置

1
conn.autocommit = True

修改数据

1
2
3
4
5
6
modify_sql = """
UPDATE `table_name` SET `name`='李小龙',`birth_date`='1940-11-27' WHERE `gender`='M'
"""
cursor.execute(modify_sql)
# 提交事务
conn.commit()

查询数据

1
2
3
4
5
6
7
8
query_sql = """
SELECT `t_id`,`name`,`gender`,`birth_date` FROM `table_name` WHERE `birth_date` BETWEEN %s and %s
"""
start_date = date(1990, 1, 1)
end_date = date(2000, 1, 1)
cursor.execute(query_sql, (start_date, end_date))
for (t_id, name, gender, birth_date) in cursor:
    print("{},{},{},{}".format(t_id, name, gender, birth_date))

删除数据

1
2
3
4
5
6
7
8
9
10
delete_sql = """
DELETE FROM `table_name` WHERE `name` = %(name)s
"""
delete_action = {
    "name": "姓名"
}
# 执行sql
cursor.execute(delete_sql, delete_action)
# 提交事务
conn.commit()

完整实例

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
import mysql.connector
from datetime import date

config = {
    "user": "root",
    "password": "123456",
    "host": "127.0.0.1",
    "database": "py_example"
}


def connect(conf):
    try:
        cnx = mysql.connector.connect(**conf)
    except mysql.connector.Error as err:
        print(err)
    else:
        return cnx


def create_database(db_cursor, db_name):
    try:
        db_cursor.execute("CREATE DATABASE {} DEFAULT CHARACTER SET 'utf8mb4'".format(db_name))
    except mysql.connector.Error as err:
        print(err.msg)


def create_tables(db_cursor, sql_str):
    try:
        db_cursor.execute(sql_str)
    except mysql.connector.Error as err:
        print(err.msg)


def execute(db_cursor, sql_str, data_obj=None):
    try:
        db_cursor.execute(sql_str, data_obj)
    except mysql.connector.Error as err:
        print(err.msg)


if __name__ == '__main__':
    # 获取连接对象
    conn = connect(config)

    # 设置自动提交事务
    conn.autocommit = True

    # 获取操作游标
    cursor = conn.cursor()

    # 创建数据库
    create_database(cursor, "db_example")

    # 创建表
    table_sql = """
    CREATE TABLE `table_example` (
      `t_id` INT(11) NOT NULL AUTO_INCREMENT,
      `name` VARCHAR(255) NOT NULL ,
      `gender` ENUM('M','F') NOT NULL ,
      `birth_date` DATE NOT NULL,
      PRIMARY KEY (`t_id`)
    ) ENGINE = INNODB
    """
    create_tables(cursor, table_sql)

    # 插入数据
    insertSQL = """
    INSERT INTO `table_example` (`name`,`gender`,`birth_date`) VALUES ('蓝士钦','M','1995-01-29')
    """
    execute(cursor, insertSQL)

    # 插入数据
    insert_sql_template = """
    INSERT INTO `table_example` (`name`,`gender`,`birth_date`) VALUES (%s,%s,%s)
    """
    insert_sql_date = ('测试姓名', 'M', date(1996, 8, 30))
    execute(cursor, insert_sql_template, insert_sql_date)

    # 插入数据
    sql = """
    INSERT INTO `table_example` (`name`,`gender`,`birth_date`) VALUES (%(name)s,%(gender)s,%(birth_date)s)
    """

    data = {
        "name": "自由女神",
        "gender": "F",
        "birth_date": date(1874, 1, 1),
    }
    execute(cursor, sql, data)

    # 修改数据
    modify_sql = """
    UPDATE `table_example` SET `name`='李小龙',`birth_date`='1940-11-27' WHERE `gender`='M'
    """
    execute(cursor, modify_sql)

    # 查询数据
    query_sql = """
    SELECT `t_id`,`name`,`gender`,`birth_date` FROM `table_example` WHERE `birth_date` BETWEEN %s and %s
    """
    start_date = date(1900, 1, 1)
    end_date = date(2000, 1, 1)
    try:
        execute(cursor, query_sql, (start_date, end_date))
        for (t_id, name, gender, birth_date) in cursor:
            print("{},{},{},{}".format(t_id, name, gender, birth_date))
    except mysql.connector.Error as err_obj:
        print(err_obj.msg)

    # 删除数据
    delete_sql = """
    DELETE FROM `table_example` WHERE `name` = %(name)s
    """
    delete_param = {
        "name": "李小龙"
    }
    execute(cursor, delete_sql, delete_param)

    # 关闭操作游标
    cursor.close()
    # 关闭数据库连接
    conn.close()
0%