PyMySQL 快速上手:用 Python 轻松玩转数据库

pyMySQL是一个纯Python编写的MySQL 客户端库，它兼容 MySQLdb1，并且易于安装和使用。本文将详细介绍如何使用PyMySQL 连接、操作MySQL数据库。

1. 安装 PyMySQL

使用 pip 安装 PyMySQL：

pip install pymysql

2. 连接数据库

在使用 PyMySQL 操作数据库之前，需要先建立与数据库的连接。以下是连接数据库的基本步骤：

import pymysql# 数据库连接参数db_config = {    'host': 'localhost',  # 数据库主机地址 如果自己的改了名称，需要改成自己的    'port': 3306,         # 数据库端口，默认为 3306    'user': 'your_user',  # 数据库用户名 要改成自己的    'password': 'your_password',  # 数据库密码 要改成自己的    'database': 'your_database',  # 要连接的数据库名 要改成自己的    'charset': 'utf8mb4'# 字符集，建议使用 utf8mb4 以支持 Emoji 表情等}try:    # 建立数据库连接    connection = pymysql.connect(**db_config)    # 创建游标对象    with connection.cursor() as cursor:        # 在此处执行 SQL 查询或操作        print("成功连接到数据库!")    # 关闭数据库连接 (with 语句块会自动处理)    # connection.close()except pymysql.MySQLError as e:    print(f"数据库连接错误: {e}")

代码解释：

• db_config 字典： 存储数据库连接所需的参数。根据你的实际情况修改这些参数。
• pymysql.connect(**db_config)： 使用 connect() 函数建立数据库连接。 **db_config 会将字典解包为关键字参数。
• connection.cursor()： 创建一个游标对象。游标用于执行 SQL 语句并获取结果。
• with connection.cursor() as cursor:： 使用 with 语句可以确保在代码块结束后自动关闭游标，即使发生异常也能保证资源被正确释放。这是一个良好的编程习惯。
• try...except 块： 捕获并处理数据库连接过程中可能出现的错误 (如用户名/密码错误、数据库不存在等)。
• connection.close(): 当你使用with语句时, 可以不需要手动关闭数据库连接.

3. 执行 SQL 语句

连接成功后，可以使用游标对象执行各种 SQL 语句，包括：

• 查询 (SELECT)
• 插入 (INSERT)
• 更新 (UPDATE)
• 删除 (DELETE)
• 创建表 (CREATE TABLE)
• 删除表 (DROP TABLE)
• ...等等

3.1 创建表 (CREATE TABLE)

with connection.cursor() as cursor:    # 创建表的 SQL 语句 (如果表不存在)    create_table_sql = """    CREATE TABLE IF NOT EXISTS your_table (        id INT AUTO_INCREMENT PRIMARY KEY,        column1 VARCHAR(255),        column2 INT,        column3 INT    )    """    cursor.execute(create_table_sql)    print("表 'your_table' 创建成功 (或已存在)")    # 提交更改 (对于 DDL 语句，如 CREATE TABLE，通常也需要提交)    connection.commit()

代码解释：

• CREATE TABLE IF NOT EXISTS your_table (...)： 创建表的 SQL 语句。
• IF NOT EXISTS 子句确保只有当表不存在时才创建，避免重复创建导致错误。
• id INT AUTO_INCREMENT PRIMARY KEY：创建一个名为 id 的整数类型列，设置为自动递增的主键。
• column1 VARCHAR(255) NOT NULL：创建一个名为 column1 的字符串类型列，最大长度为 255，且不能为空。
• column2 INT：创建一个名为 column2 的整数类型列。
• column3 DATE：创建一个名为 column3 的日期类型列。
• cursor.execute(create_table_sql): 执行创建表的语句
• connection.commit(): 提交更改。虽然 CREATE TABLE 通常会自动提交，但显式调用 commit() 是一个好习惯，可以确保更改被保存。

3.2 插入数据 (INSERT)

with connection.cursor() as cursor:    # 执行 SQL 插入    sql = "INSERT INTO your_table (column1,column2,column3) VALUES (%s,%s,%s)"    data = [(11, 12,12333), (21, 22,22222), (1,2,3)]                cursor.executemany(sql, data)    # 提交事务 (重要!)    connection.commit()    print(f"{cursor.rowcount} 行数据插入成功!")  # cursor.rowcount 返回受影响的行数

我们可以看看数据：

代码解释：

• cursor.executemany(sql,data): 执行了插入多行数据的SQL语句
• connection.commit(): 非常重要! 对于修改数据的操作 (INSERT, UPDATE, DELETE)，必须调用 connection.commit() 来提交事务，否则更改不会保存到数据库中。

3.3 查询数据 (SELECT)

with connection.cursor() as cursor:    # 执行 SQL 查询    sql = "SELECT * FROM your_table WHERE column1 = %s"    cursor.execute(sql, (11,))  # 使用参数化查询，防止 SQL 注入    # 获取查询结果    # 1. 获取所有结果    results = cursor.fetchall()    for row in results:        print(row)  # row 是一个元组，对应一行数据    # 2. 获取一行结果    # result = cursor.fetchone()    # print(result)    # 3. 获取指定数量的结果    # results = cursor.fetchmany(5)  # 获取前 5 行    # print(results)

• cursor.execute(sql, params)： 执行 SQL 查询。
• sql: 要执行的 SQL 语句。
• params: (可选) 一个元组或列表，包含要替换 SQL 语句中占位符 (%s) 的值。强烈建议使用参数化查询，这可以防止 SQL 注入攻击，并提高安全性。
• cursor.fetchall()： 获取所有查询结果。返回一个包含元组的列表，每个元组代表一行数据。
• cursor.fetchone()： 获取下一行查询结果。如果没有更多行，则返回 None。
• cursor.fetchmany(size)： 获取指定数量的查询结果。

3.4 更新数据 (UPDATE)

with connection.cursor() as cursor:    # 执行 SQL 更新    sql = "UPDATE your_table SET column2 = %s WHERE column1 = %s"    data = (1000, 11)#把column1列数据为11的行其对应的column2更新为1000    cursor.execute(sql, data)    # 提交事务    connection.commit()    print(f"{cursor.rowcount} 行数据更新成功!")

我们看看数据库更新完后的结果。

3.5 删除数据 (DELETE)

with connection.cursor() as cursor:    # 执行 SQL 删除    sql = "DELETE FROM your_table WHERE column1 = %s"    data = (11,)    cursor.execute(sql, data)    # 提交事务    connection.commit()    print(f"{cursor.rowcount} 行数据删除成功!")

3.6 使用字典游标 (DictCursor)

默认情况下，cursor.fetchall() 返回的结果是元组的列表。如果希望以字典的形式获取结果（键是列名，值是对应的数据），可以使用 DictCursor：

import pymysql.cursors# 创建游标对象connection = pymysql.connect(**db_config)with connection.cursor(pymysql.cursors.DictCursor) as cursor:    sql = "SELECT * FROM your_table"    cursor.execute(sql)    results = cursor.fetchall()    for row in results:        print(row['column1'], row['column2'])  # 通过列名访问数据

需要注意的是, connection需要重新声明

# 创建游标对象connection = pymysql.connect(**db_config)with connection.cursor(pymysql.cursors.DictCursor) as cursor:   #...

4. 事务处理

在数据库操作中，事务用于确保一组操作要么全部成功，要么全部失败，以保持数据的一致性。

try:    with connection.cursor() as cursor:        # 开始事务        connection.begin()        # 执行多个 SQL 操作        cursor.execute("INSERT ...")        cursor.execute("UPDATE ...")        cursor.execute("DELETE ...")        # 提交事务 (如果所有操作都成功)        connection.commit()except pymysql.MySQLError as e:    # 回滚事务 (如果发生错误)    connection.rollback()    print(f"事务执行失败: {e}")

解释:

• connection.begin(): 显式开始一个事务。虽然在上面的 INSERT, UPDATE, DELETE 示例中没有显式调用，但 PyMySQL 默认会为每个连接开启一个事务。connection.commit() 实际上提交的是这个隐式事务。
• connection.commit()： 提交事务。如果事务中的所有操作都成功执行，则将更改永久保存到数据库。
• connection.rollback()： 回滚事务。如果在事务执行过程中发生错误，则撤销所有已执行的操作，将数据库恢复到事务开始之前的状态。

5. 预防 SQL 注入

SQL 注入是一种常见的安全漏洞，攻击者可以通过构造恶意的输入来执行未经授权的 SQL 代码。务必使用参数化查询来防止 SQL 注入。

错误示例 (易受 SQL 注入攻击):

# 危险! 不要这样做!user_input = input("请输入用户名: ")sql = f"SELECT * FROM users WHERE username = '{user_input}'"  # 直接拼接字符串cursor.execute(sql)

如果用户输入 '; DROP TABLE users; --， 那么最终执行的 SQL 语句将变成：

SELECT * FROM users WHERE username = ''; DROP TABLE users; --'

这将导致 users 表被删除！

正确示例 (使用参数化查询):

user_input = input("请输入用户名: ")sql = "SELECT * FROM users WHERE username = %s"cursor.execute(sql, (user_input,))  # 使用参数化查询，将用户输入作为参数传递

使用参数化查询时，PyMySQL 会负责对参数进行正确的转义和处理，确保 SQL 语句的安全性。

6. 完整示例举例 (创建表、插入、查询)

import pymysqldb_config = {    'host': 'your_host',    'port': 3306,    'user': 'your_user',    'password': 'your_password',    'database': 'your_database',    'charset': 'utf8mb4'}try:    connection = pymysql.connect(**db_config)    with connection.cursor() as cursor:        # 创建表 (如果不存在)        create_table_sql = """        CREATE TABLE IF NOT EXISTS employees (            id INT AUTO_INCREMENT PRIMARY KEY,            name VARCHAR(255) NOT NULL,            age INT,            department VARCHAR(255)        )        """        cursor.execute(create_table_sql)        print("表 'employees' 创建成功 (或已存在)")        # 插入数据        insert_data = [            ('Alice', 30, 'Sales'),            ('Bob', 25, 'Marketing'),            ('Charlie', 35, 'Engineering')        ]        insert_sql = "INSERT INTO employees (name, age, department) VALUES (%s, %s, %s)"        cursor.executemany(insert_sql, insert_data)  # executemany 用于批量插入        connection.commit()        print(f"{cursor.rowcount} 行数据插入成功!")        # 查询数据        select_sql = "SELECT * FROM employees"        cursor.execute(select_sql)        results = cursor.fetchall()        print("\n查询结果:")        for row in results:            print(row)except pymysql.MySQLError as e:    print(f"数据库操作错误: {e}")finally:if connection:    connection.close()

这个示例演示了如何创建表、批量插入数据以及查询数据。cursor.executemany() 方法用于一次性插入多行数据，这比循环执行多次 cursor.execute() 更高效。 finally 块确保无论是否发生异常，数据库连接都会被关闭。

总结

PyMySQL 提供了一套简单易用的 API 来操作 MySQL 数据库。通过本文的学习，你应该掌握了以下内容：

• 安装 PyMySQL
• 连接数据库
• 执行各种 SQL 语句 (SELECT, INSERT, UPDATE, DELETE, CREATE TABLE)
• 使用参数化查询防止 SQL 注入
• 获取查询结果 (fetchall, fetchone, fetchmany)
• 使用 DictCursor 获取字典形式的结果
• 事务处理 (commit, rollback)
• 批量插入

希望这篇文章能帮助大家更好地理解和使用 PyMySQL 操作 MySQL 数据库。