Python操作MySQL数据库

发布于：2025-05-25 ⋅ 阅读:(27) ⋅ 点赞:(0)

目录

一、前言

二、安装 Python MySQL 连接库

（一）官方推荐库：mysql-connector-python

（二）替代库：PyMySQL

三、Python 连接 MySQL 数据库

（一）导入连接库

（二）创建数据库连接

（三）创建游标对象

（四）执行 SQL 语句

（五）获取查询结果

（六）关闭连接

四、常见的 MySQL 操作

（一）插入数据（INSERT）

（二）更新数据（UPDATE）

（三）删除数据（DELETE）

（四）查询数据（SELECT）

（五）执行多条 SQL 语句

（六）使用 LIKE 进行模糊查询

（七）使用 JOIN 进行联合查询

五、使用连接池

（一）连接池简介

（二）创建连接池

（三）获取连接

（四）连接池的优势

六、事务管理

（一）开始事务

（二）提交事务

（三）回滚事务

（四）事务的隔离级别

（五）事务隔离级别总结

（六）案例

七、数据库连接参数详解

八、SQL 注入攻击原理与防御

（一）攻击原理

（二）防御方法

九、游标对象的高级用法

（一）批量操作优化

（二）获取受影67响行数

十、连接池深度解析

（一）核心参数说明

（二）连接池与1011线程安全

十一、事务的原子性与一致性案例

（一）转账场景模拟

（二）事务保证原子性

十二、四种事务隔离级别实战对比

（一）准备测试表

（二）测试场景：两个事务并发执行

（三）不同隔离级别下的结果

（四）隔离级别选择建议

十三、错误处理1220最佳实践

十四、性能优化总结

（一）连接管理层面

（二）SQL 语句层面

（三）事务层面6

十七、文档关键20知识点速查表

一、前言

在现代应用程序中，数据库起着至关重要的作用。MySQL 作为流行的关系型数据库管理系统，被广泛应用于各种规模的应用。在 Python 中，可通过连接库与 MySQL 数据库交互，实现数据增、删、改、查操作。连接池作为优化数据库连接管理的技术，能提高应用性能。本文将介绍如何用 Python 操作 MySQL 数据库，包括建立连接、执行 SQL 语句、事务管理、错误处理等，重点介绍连接池的使用以优化连接效率。

二、安装 Python MySQL 连接库

（一）官方推荐库：mysql-connector-python

命令：pip install mysql-connector-python

（二）替代库：PyMySQL

命令：pip install pymysql

本文以 PyMySQL 为例进行讲解。

三、Python 连接 MySQL 数据库

（一）导入连接库

代码：import pymysql

（二）创建数据库连接

使用pymysql.connect()方法建立连接，需提供 MySQL 服务器地址、用户名、密码和数据库名。
示例代码：

db = pymysql.connect(
    host="localhost",  # MySQL数据库地址，通常为localhost或IP地址
    user="root",       # 数据库用户名，通常为root
    password="password",  # 数据库密码
    database="testdb"  # 要连接的数据库名称
)

（三）创建游标对象

建立连接后，创建游标对象以执行 SQL 语句。
代码：cursor = db.cursor()

（四）执行 SQL 语句

通过游标对象的execute()方法执行 SQL 语句，使用%s占位避免 SQL 注入攻击。
示例：cursor.execute("SELECT * FROM users")

（五）获取查询结果

  • fetchall()：获取所有结果。
  • fetchone()：获取单条记录。
    示例代码：
results = cursor.fetchall()
for row in results:
    print(row)

（六）关闭连接

操作完成后，关闭游标和数据库连接。
代码：

cursor.close()
db.close()

四、常见的 MySQL 操作

（一）插入数据（INSERT）

使用INSERT INTO语句，通过execute()方法执行，值用%s占位符。
示例代码：

cursor.execute("INSERT INTO users (name, age) VALUES (%s, %s)", ("Alice", 25))
db.commit()  # 提交事务，保存插入数据

（二）更新数据（UPDATE）

使用UPDATE语句，通过execute()方法执行，添加WHERE条件确保更新特定记录。
示例代码：

cursor.execute("UPDATE users SET age = %s WHERE name = %s", (26, "Alice"))
db.commit()  # 提交事务，保存更新数据

（三）删除数据（DELETE）

使用DELETE语句，通过WHERE条件确保删除特定记录。
示例代码：

cursor.execute("DELETE FROM users WHERE name = %s", ("Alice",))
db.commit()  # 提交事务，保存删除数据

（四）查询数据（SELECT）

使用SELECT语句，用fetchall()或fetchone()获取记录。
示例代码：

cursor.execute("SELECT * FROM users")
results = cursor.fetchall()
for row in results:
    print(row)

（五）执行多条 SQL 语句

对于批量插入、更新等操作，使用executemany()方法一次执行多条 SQL 语句。
示例代码：

cursor.executemany(
    "INSERT INTO users(name, age) VALUES(%s, %s)",
    [("Bob", 30), ("Charlie", 35), ("David", 28)]
)
db.commit()  # 提交事务

（六）使用 LIKE 进行模糊查询

使用LIKE关键字和%通配符进行模糊查询。
示例代码：

cursor.execute("SELECT * FROM users WHERE name LIKE %s", ("%a%",))
results = cursor.fetchall()
for row in results:
    print(row)

（七）使用 JOIN 进行联合查询

使用JOIN关键字合并多个表的数据。
示例代码：

cursor.execute(
    "SELECT users.name, orders.amount "
    "FROM users "
    "INNER JOIN orders ON users.id = orders.user_id"
)
results = cursor.fetchall()
for row in results:
    print(row)

五、使用连接池

（一）连接池简介

连接池技术在高并发场景下可提升数据库连接效率。池中提前创建多个数据库连接，客户端从池获取连接，减少连接创建和销毁的开销。

（二）创建连接池

PyMySQL 不直接支持连接池，使用 DBUtils 库创建，需先安装 DBUtils。
命令：pip install dbutils
示例代码：

from dbutils.pooled_db import PooledDB
import pymysql

# 数据库连接配置
dbconfig = {
    "host": "localhost",
    "user": "root",
    "password": "password",
    "database": "testdb"
}

# 创建连接池
connection_pool = PooledDB(
    creator=pymysql,  # 使用PyMySQL作为数据库连接库
    maxconnections=5,  # 连接池中最大连接数
    **dbconfig
)

（三）获取连接

从连接池获取连接，使用connection()方法，操作完成后连接自动归还池。
示例代码：

db_connection = connection_pool.connection()
cursor = db_connection.cursor()
cursor.execute("SELECT * FROM users")
results = cursor.fetchall()
for row in results:
    print(row)
cursor.close()
db_connection.close()  # 连接自动归还给连接池

（四）连接池的优势

  1. 性能提升：减少创建新连接的开销，提高数据库操作效率。
  2. 资源管理：限制最大连接数，避免数据库过载。
  3. 更易管理：统一管理连接生命周期，简化代码结构。

六、事务管理

事务是由多个 SQL 语句组成的工作单元，保证数据原子性，即所有操作要么都成功，要么都失败。

（一）开始事务

可通过START TRANSACTION显式开启，一般执行 SQL 语句启动事务。
示例：cursor.execute("START TRANSACTION")

（二）提交事务

若事务中所有操作成功，使用commit()方法提交事务，保存更改。
代码：db.commit()

（三）回滚事务

若事务中操作失败，使用rollback()方法回滚事务，撤销更改。
代码：db.rollback()

（四）事务的隔离级别

MySQL 支持四种事务隔离级别，定义了并发事务间的影响，隔离级别越高，干扰越小，性能可能下降，默认隔离级别为REPEATABLE READ。

  1. READ UNCOMMITTED（未提交读）
    • 描述：事务可读取其他事务未提交的数据，可能导致 “脏读”，隔离性最差。
    • 应用场景：不推荐，除非对数据一致性要求不高。
    • 使用方法：cursor.execute("SET TRANSACTION ISOLATION LEVEL READ UNCOMMITTED")
  2. READ COMMITTED（提交读）
    • 描述：事务只能读取其他事务已提交的数据，避免脏读，可能遇到 “不可重复读”，即两次读取数据可能变化（另一事务已提交修改）。
    • 应用场景：适用于大多数常见场景，提供一定隔离性，保证较好性能。
    • 使用方法：cursor.execute("SET TRANSACTION ISOLATION LEVEL READ COMMITTED")
  3. REPEATABLE READ（可重复读）
    • 描述：事务保证在事务内多次读取同一数据时值不变，避免 “不可重复读”，但可能出现 “幻读”，即读取的数据集在事务执行中变化。
    • 应用场景：需要保证事务数据一致性的场景，如金融系统余额操作等。
    • 使用方法：cursor.execute("SET TRANSACTION ISOLATION LEVEL REPEATABLE READ")
  4. SERIALIZABLE（串行化）
    • 描述：最严格的隔离级别，事务像串行一样执行，完全避免脏读、不可重复读和幻读，性能开销最大，可能导致大量锁竞争。
    • 应用场景：对数据一致性要求极高的场景，如库存管理系统、银行转账等。
    • 使用方法：cursor.execute("SET TRANSACTION ISOLATION LEVEL SERIALIZABLE")

（五）事务隔离级别总结

隔离级别 是否允许脏读 是否解决不可重复读 是否解决幻读 性能 应用场景
READ UNCOMMITTED 允许 否 否 最好 数据一致性要求不高
READ COMMITTED 否 否 否 较好 大多数常见场景
REPEATABLE READ 否 是 否 一般 数据一致性要求较高场景（默认）
SERIALIZABLE 否 是 是 最差 数据一致性要求极高场景

选择隔离级别时，需根据应用需求平衡数据一致性和性能。数据不频繁冲突时，选较低级别提升性能；数据一致性要求极高时，选较高级别，尽管性能有损失。

（六）案例

import pymysql

# 连接到数据库
conn = pymysql.connect(
    host='localhost',
    user='root',
    password='password',
    database='testdb'
)

# 创建游标对象
cursor = conn.cursor()

# 关闭自动提交模式
conn.autocommit = False

try:
    # 开始事务
    # 执行插入操作
    cursor.execute("INSERT INTO users (name, age) VALUES ('Eve', 22)")
    cursor.execute("INSERT INTO orders (user_id, amount) VALUES ((SELECT id FROM users WHERE name = 'Eve'), 120.58)")
    
    # 提交事务
    conn.commit()
    print("事务已提交。")
except pymysql.MySQLError as err:
    # 发生错误，回滚事务
    print(f"错误: {err}")
    conn.rollback()
    print("事务已回滚。")
finally:
    # 关闭游标和连接
    cursor.close()
    conn.close()

七、数据库连接参数详解

在使用pymysql.connect()建立连接时，除基础参数外，还有一些重要配置可优化连接行为：

  • port：指定 MySQL 服务器端口，默认值为3306，示例：port=3307。
  • charset：设置连接字符集，避免中文乱码问题，常用值为utf8mb4（支持 Emoji 等特殊字符），示例：charset="utf8mb4"。
  • autocomm{insert\_element\_0\_}it：是否开启自动提交模式，默认值为False。若设为True，每次执行 SQL 后会自动提交事务，无需手动调用commit()，但可能降低事务安全性。
  • cursorcl{insert\_element\_1\_}ass：自定义游标类型，如使用pymysql.cursors.DictCursor可返回字典形式结果，便于字段访问： 
    cursor = db.cursor(cursorclass=pymysql.cursors.DictCursor)
results = cursor.fetchall()  # 结果为[{'name': 'Alice', 'age': 25}, ...]

八、SQL 注入攻击原理与防御

（一）攻击原理

通过在 SQL 语句中插入恶意代码，篡改原本逻辑。例如，用户输入"'; DROP TABLE users; --"作为用户名时，若直接拼接 SQL 会导致：

# 不安全写法（易受攻击）
username = "恶意输入"
cursor.execute("SELECT * FROM users WHERE name = '" + username + "'")
# 拼接后语句：SELECT * FROM users WHERE name = ''; DROP TABLE users; --'

执行时会先闭合单引号，再执行DROP TABLE删除表，--注释后续内容。

（二）防御方法

  1. 使用占位符%s：

    • 所有动态参数通过%s占位，由驱动层自动转义，避免拼接风险。
    • 正确写法：

      python

      username = "恶意输入"
cursor.execute("SELECT * FROM users WHERE name = %s", (username,))  # 自动转义特殊字符
    • 原理：数据库驱动会将%s视为独立参数，与 SQL 语句分离处理，阻断恶意代码注入路径。

  2. 限制输入长度与45格式：

    • 对用户输入进行正则校验（如手机号、邮箱格式），拒绝非法字符。
    • 使用框架自带的 ORM（对象关系映射）工具（如 Django ORM），其内部已封装安全的参数传递机制。

九、游标对象的高级用法

（一）批量操作优化

当插入大量数据时，executemany()比循环执行execute()效率更高，减少与数据库的交互次数：

# 批量插入1000条数据
data = [("User%d" % i, 18) for i in range(1000)]
cursor.executemany("INSERT INTO users (name, age) VALUES (%s, %s)", data)
db.commit()

  • 优势：仅需一次网络传输即可发送多条语句，适用于 CSV 导入、日志批量写入等场景。
（二）获取受影67响行数

通过cursor.rowcount属性获取最近一次执行 SQL 影响的行数，常用于判断更新 / 删除是否成功：

cursor.execute("DELETE FROM users WHERE age < 18")
print(f"删除了{cursor.rowcount}条数据")  # 输出：删除了5条数据

十、连接池深度解析

（一）核心参数说明

在使用DBUtils.PooledDB创建连接池时，可配置以下参数优化性能：

  • mincached：连接池中初始化的空闲连接数，默认值为0。若设为2，启动时会创建 2 个空闲连接等待使用，减少首次请求的连接创建延迟。
  • maxcached：连接池中最多可保持的空闲连接数，默认值为0（不限制）。若设为5，当空闲连接超过 5 个时，会自动关闭多余连接，释放资源。
  • maxusage：单个连接的最大使用次数，默认值为0（不限制）。达到次数后，连接会被强制关闭并重新创建，避免长时间使用导致的连接泄漏。
（二）连接池与1011线程安全
  • PooledDB基于线程本地存储（Thread Local Storage）实现，每个线程获取的连接独立，不会出现线程间连接混用的问题。
  • 示例：多线程环境下安全使用连接池： 
    import threading
from dbutils.pooled_db import PooledDB
import pymysql

pool = PooledDB(creator=pymysql, maxconnections=10, ...)

def query_data():
    conn = pool.connection()
    cursor = conn.cursor()
    cursor.execute("SELECT COUNT(*) FROM users")
    result = cursor.fetchone()
    conn.close()  # 连接归还池中，自动释放给其他线程
    print(f"用户数：{result[0]}")

# 启动10个线程并发查询
threads = [threading.Thread(target=query_data) for _ in range(10)]
for t in threads: t.start()
for t in threads: t.join()

十一、事务的原子性与一致性案例

（一）转账场景模拟

假设用户 A 向用户 B 转账 100 元，需执行以下步骤：

  1. 扣除用户 A 的余额（UPDATE accounts SET balance = balance - 100 WHERE name = 'A'）。
  2. 增加用户 B 的余额（UPDATE accounts SET balance = balance + 100 WHERE name = 'B'）。
（二）事务保证原子性

若步骤 1 执行成功后系统崩溃，未执行步骤 2，事务未提交时，步骤 1 的修改会被回滚，确保数据不丢失、不脏写：

try:
    conn.autocommit = False  # 关闭自动提交，开启事务
    # 扣除A的余额
    cursor.execute("UPDATE accounts SET balance = balance - 100 WHERE name = 'A'")
    # 模拟系统崩溃（抛出异常）
    raise Exception("系统错误")
    # 增加B的余额（未执行）
    cursor.execute("UPDATE accounts SET balance = balance + 100 WHERE name = 'B'")
    conn.commit()  # 不会执行到这里
except:
    conn.rollback()  # 回滚，A的余额恢复原状
finally:
    conn.close()

十二、四种事务隔离级别实战对比

（一）准备测试表
CREATE TABLE test_transaction (
    id INT PRIMARY KEY AUTO_INCREMENT,
    value INT
);
INSERT INTO test_transaction (value) VALUES (100);
（二）测试场景：两个事务并发执行
事务 1（会话 1） 事务 2（会话 2）
START TRANSACTION; START TRANSACTION;
SELECT value FROM test_transaction WHERE id = 1;（读取值为 100） UPDATE test_transaction SET value = 200 WHERE id = 1;
-- 等待事务2执行操作 -- COMMIT;
SELECT value FROM test_transaction WHERE id = 1; -- 无操作 --
（三）不同隔离级别下的结果
  1. READ UNCOMMITTED：
    • 事务 1 第二次读取值为 200（读取到事务 2 未提交的修改，脏读）。
  2. READ COMMITTED：
    • 事务 1 第二次读取值为 200（事务 2 已提交，可重复读不成立，出现不可重复读）。
  3. REPEATABLE READ（MySQL 默认）：
    • 事务 1 第二次读取值仍为 100（通过 MVCC 机制实现可重复读，避免不可重复读）。
  4. SERIALIZABLE：
    • 事务 2 的UPDATE会被阻塞，直到事务 1 提交或回滚，最终事务 1 读取值为 200（串行执行，无幻读）。
（四）隔离级别选择建议
  • 互联网应用：通常使用READ COMMITTED（如 Oracle 默认级别），在保证一致性的同时提供较高并发性能。
  • 金融交易系统：选择REPEATABLE READ或SERIALIZABLE，确保资金操作的严格一致性，宁可牺牲部分性能。

十三、错误处理1220最佳实践

在数据库操作中，需捕获常见异常并处理，避免程序崩溃：

try:
    # 执行可能出错的SQL
    cursor.execute("INSERT INTO users (name) VALUES (%s)", ("Alice",))
    db.commit()
except pymysql.MySQLError as err:
    # 捕获数据库异常
    error_code = err.args[0]
    error_msg = err.args[1]
    if error_code == 1062:  # 唯一键冲突（Duplicate entry）
        print(f"错误：重复数据，原因：{error_msg}")
    elif error_code == 1045:  # 权限拒绝（Access denied）
        print("错误：数据库连接权限不足")
    else:
        print(f"未知错误：{error_msg}")
    db.rollback()  # 回滚事务
finally:
    # 确保资源释放
    if 'cursor' in locals() and not cursor.closed:
        cursor.close()
    if 'db' in locals() and db.open:
        db.close()

  • 常见错误码：
    • 1062：唯一键冲突（如重复插入相同用户名）。
    • 1451：外键约束冲突（删除主表数据前未删除子表相关记录）。
    • 2003：无法连接数据库（服务器地址错误或端口被防火墙阻断）。

十四、性能优化总结

（一）连接管理层面
  • 使用连接池替代每次请求创建新连接，减少 TCP 握手和认证开销。
  • 合理设置连接池max{insert\_element\_6\_}connections参数，避免超过 MySQL 服务器的max_connections限制（可通过SHOW VARIABLES LIKE 'max_connections'查看）。
（二）SQL 语句层面
  • 对高频查询字段添加索引，避免全表扫描（如CREATE INDEX idx_name ON users(name)）。
  • 避免使用SELECT *，只查询需要的字段，减少数据传输量。
  • 批量操作时使用executemany()而非循环execute()，提升插入 / 更新效率。
（三）事务层面6
  • 尽量缩短事务范围，将非必要操作移出事务块，减少锁持有时间。
  • 对读多写少的场景，可降低隔离级别（如使用READ COMMITTED）以提升并发度。

十七、文档关键20知识点速查表

知识点 核心内容 文档对应段落
连接 MySQL pymysql.connect()参数：host、user、password、database
执行 SQL c{insert\_element\_9\_}ursor.execute()用于单条语句，executemany()用于批量操作
事务管理 co{insert\_element\_10\_}mmit()提交，rollback()回滚，默认隔离级别REPEATABLE READ
连接池 使用D{insert\_element\_11\_}BUtils.PooledDB创建，参数maxconnections控制最大连接数
隔离级别 四种级911别：READ UNCOMMITTED、READ COMMITTED、REPEATABLE READ、SERIALIZABLE

网站公告

今日签到

点亮在社区的每一天
去签到

热门文章

最新发布