Python操作MySQL数据库实践

更新时间：2026年04月11日 08:55:16 作者：kiku1818

本文详细介绍了Python操作MySQL数据库的全流程技术,从基础连接库安装、CRUD操作,到进阶的连接池优化、事务管理与隔离级别,并补充了异常处理、SQL注入防护、批量操作优化等知识点,帮助读者掌握Python操作MySQL的核心技能,适用于运维开发及数据库学习者

本文系统梳理 Python 操作 MySQL 数据库的全流程技术栈，从基础的 PyMySQL 连接库安装、数据库连接与增删改查操作，到进阶的连接池优化、事务管理与隔离级别深度解析，同时补充了图片未覆盖的异常处理、SQL 注入防护、批量操作优化、上下文管理器等实用知识点。

全文以运维开发视角，结合实战代码与原理说明，帮助读者掌握 Python 操作 MySQL 的核心技能，解决高并发场景下的性能瓶颈与数据一致性问题，适合运维工程师、后端开发及数据库学习者参考学习。

一、前言

在现代应用程序与运维自动化开发中，数据库操作是核心环节之一。

MySQL 作为最流行的开源关系型数据库管理系统（RDBMS），广泛应用于中小规模业务系统、运维监控平台、数据采集工具等场景。

Python 凭借简洁的语法和丰富的生态，成为操作 MySQL 的首选语言之一：我们可以通过连接库与 MySQL 数据库交互，实现数据的增、删、改、查，同时通过连接池、事务管理等技术优化性能、保障数据一致性。

二、Python MySQL 连接库的选择与安装

2.1 主流连接库对比

Python 操作 MySQL 有多个成熟的连接库，不同库的适用场景和特性各有差异：

连接库	特点	适用场景
mysql-connector-python	MySQL 官方推荐库，纯 Python 实现，无需依赖 C 编译器，兼容性强	跨平台部署、无编译环境的场景
PyMySQL	纯 Python 实现，兼容 MySQLdb 接口，轻量易用，社区活跃	主流 Python 项目、运维自动化脚本
mysqlclient	基于 C 语言实现，性能更高，兼容 Django 等框架	高并发、高性能要求的后端项目

图片中重点介绍了mysql-connector-python和PyMySQL，其中PyMySQL是目前 Python 生态中最常用的选择，本文后续示例均以 PyMySQL 为例。

2.2 安装命令

（1）安装 mysql-connector-python

运行

pip install mysql-connector-python

（2）安装 PyMySQL（推荐）

运行

pip install pymysql

（3）补充：安装高性能 mysqlclient

运行

# Windows
pip install mysqlclient
# Linux (Ubuntu/Debian)
sudo apt-get install python3-dev default-libmysqlclient-dev build-essential
pip install mysqlclient
# CentOS/RHEL
sudo yum install python3-devel mysql-devel
pip install mysqlclient

三、Python 连接 MySQL 数据库的完整流程

3.1 基础连接步骤

（1）导入连接库

运行

import pymysql

（2）创建数据库连接

使用pymysql.connect()方法建立连接，核心参数说明：

host：MySQL 服务器地址，本地为localhost，远程为 IP 地址
user：数据库用户名，通常为root
password：数据库密码
database：要连接的数据库名称
port：MySQL 端口，默认 3306（可省略）
charset：字符集，推荐utf8mb4（支持 emoji 等特殊字符）
connect_timeout：连接超时时间（秒），避免长时间等待

完整示例：

运行

# 建立数据库连接
db = pymysql.connect(
    host="localhost",
    user="root",
    password="your_password",
    database="testdb",
    port=3306,
    charset="utf8mb4",
    connect_timeout=10
)

（3）创建游标对象

游标是执行 SQL 语句的核心对象，用于发送 SQL 命令、接收执行结果：

运行

cursor = db.cursor()

补充知识点：游标类型

默认游标：返回元组格式的结果
DictCursor：返回字典格式的结果，键为列名，更便于数据处理

运行

from pymysql.cursors import DictCursor
cursor = db.cursor(cursor=DictCursor)

（4）执行 SQL 语句

通过游标execute()方法执行 SQL，支持%s占位符防 SQL 注入：

运行

# 查询示例
cursor.execute("SELECT * FROM users")

（5）获取查询结果

fetchall()：获取所有查询结果，返回二维元组 / 列表
fetchone()：获取单条查询结果，返回一维元组 / 字典
fetchmany(size)：获取指定数量的结果

示例：

运行

# 获取所有结果
results = cursor.fetchall()
for row in results:
    print(row)

# 获取单条结果
one_row = cursor.fetchone()
print(one_row)

（6）关闭连接

操作完成后必须关闭游标和连接，避免资源泄漏：

运行

cursor.close()
db.close()

3.2 补充：上下文管理器优化连接

使用with语句自动管理连接和游标，无需手动关闭，避免资源泄漏：

运行

# 连接上下文管理器
with pymysql.connect(
    host="localhost",
    user="root",
    password="your_password",
    database="testdb",
    charset="utf8mb4"
) as db:
    # 游标上下文管理器
    with db.cursor() as cursor:
        cursor.execute("SELECT * FROM users")
        results = cursor.fetchall()
        for row in results:
            print(row)
# 自动关闭连接和游标

四、MySQL 常见 CRUD 操作详解

4.1 插入数据（INSERT）

使用INSERT INTO语句，通过%s占位符防 SQL 注入，执行后需提交事务：

运行

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

# 批量插入（executemany）
sql = "INSERT INTO users (name, age) VALUES (%s, %s)"
data = [("Bob", 30), ("Charlie", 35), ("David", 28)]
cursor.executemany(sql, data)
db.commit()

补充知识点：批量插入优化

executemany()比多次execute()性能提升 10-100 倍，适合批量数据导入
可通过cursor.lastrowid获取自增主键 ID：print(cursor.lastrowid)

4.2 更新数据（UPDATE）

使用UPDATE语句，必须加WHERE条件，避免全表更新：

运行

sql = "UPDATE users SET age = %s WHERE name = %s"
cursor.execute(sql, (26, "Alice"))
db.commit()

4.3 删除数据（DELETE）

使用DELETE语句，必须加WHERE条件，避免误删全表数据：

运行

sql = "DELETE FROM users WHERE name = %s"
cursor.execute(sql, ("Alice",))  # 注意逗号，确保是元组
db.commit()

4.4 查询数据（SELECT）

基础查询、条件查询、模糊查询、联合查询：

（1）基础查询

运行

cursor.execute("SELECT * FROM users")
results = cursor.fetchall()

（2）条件查询

运行

sql = "SELECT * FROM users WHERE age > %s"
cursor.execute(sql, (25,))
results = cursor.fetchall()

（3）模糊查询（LIKE）

运行

# 查询名字包含"a"的用户
sql = "SELECT * FROM users WHERE name LIKE %s"
cursor.execute(sql, ("%a%",))
results = cursor.fetchall()

（4）联合查询（JOIN）

多表关联查询，常用INNER JOIN、LEFT JOIN：

运行

sql = """
SELECT users.name, orders.amount
FROM users
INNER JOIN orders ON users.id = orders.user_id
"""
cursor.execute(sql)
results = cursor.fetchall()

4.5 补充：SQL 注入防护

核心原则：永远不要用字符串拼接 SQL 语句，必须使用%s占位符

错误示例（存在注入风险）：

运行

# 危险！用户输入恶意内容会导致SQL注入
sql = f"SELECT * FROM users WHERE name = '{user_input}'"

正确示例（占位符防注入）：

运行

sql = "SELECT * FROM users WHERE name = %s"
cursor.execute(sql, (user_input,))

补充防护：使用预编译语句、限制数据库用户权限、开启防火墙等

五、连接池技术：高并发场景的性能优化

5.1 连接池的核心原理

直接创建数据库连接存在严重的性能问题：每次连接都需要 TCP 三次握手、身份验证、资源分配，频繁创建 / 销毁会导致系统性能急剧下降。

连接池的核心思想：提前创建一批数据库连接，放入连接池中，客户端需要时从池中获取，使用后归还，避免重复创建 / 销毁的开销，同时限制最大连接数，防止数据库过载。

5.2 基于 DBUtils 实现连接池

PyMySQL 本身不支持连接池，需使用DBUtils库实现，安装命令：

运行

pip install dbutils

完整连接池实现代码

运行

from dbutils.pooled_db import PooledDB
import pymysql

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

# 创建连接池
connection_pool = PooledDB(
    creator=pymysql,  # 使用PyMySQL作为连接库
    maxconnections=5,  # 最大连接数，根据数据库配置调整
    mincached=2,  # 初始化时创建的空闲连接数
    maxcached=3,  # 最大空闲连接数
    maxshared=3,  # 最大共享连接数
    blocking=True,  # 连接池满时是否阻塞等待
    setsession=[],  # 会话初始化SQL
    ping=1,  # 自动检测连接有效性（1=每次获取连接时检测）
    **dbconfig
)

# 从连接池获取连接
db_connection = connection_pool.connection()
cursor = db_connection.cursor()

# 执行SQL操作
cursor.execute("SELECT * FROM users")
results = cursor.fetchall()
for row in results:
    print(row)

# 关闭游标和连接（连接自动归还到连接池）
cursor.close()
db_connection.close()

5.3 连接池的核心优势

性能提升：减少连接创建 / 销毁的开销，高并发场景下性能提升 5-10 倍
资源管理：限制最大连接数，避免过多连接导致数据库崩溃
高可用性：自动检测连接有效性，避免使用失效连接
统一管理：统一配置连接参数，简化代码维护

5.4 补充：连接池参数详解

maxconnections：最大连接数，建议设置为 MySQLmax_connections的 80% 以内
mincached：最小空闲连接数，保证随时有可用连接
maxcached：最大空闲连接数，避免空闲连接占用过多资源
ping：连接有效性检测，1表示每次获取连接时检测，防止 "僵死连接"
blocking：连接池满时，True表示阻塞等待，False表示抛出异常

六、事务管理：保障数据一致性

6.1 事务的核心特性（ACID）

事务是由多个 SQL 语句组成的工作单元，必须满足 ACID 特性：

原子性（Atomicity）：事务中的操作要么全部成功，要么全部失败回滚
一致性（Consistency）：事务执行前后，数据库的完整性约束不被破坏
隔离性（Isolation）：多个并发事务之间相互隔离，互不干扰
持久性（Durability）：事务提交后，数据永久保存到数据库，不会因系统故障丢失

6.2 事务的基本操作

（1）开启事务

MySQL 默认自动提交事务，需手动关闭自动提交，开启事务：

运行

# 关闭自动提交（开启事务模式）
db.autocommit = False
# 或显式开启事务
cursor.execute("START TRANSACTION")

（2）提交事务

所有操作成功后，提交事务，保存数据：

运行

db.commit()

（3）回滚事务

操作失败时，回滚事务，撤销所有修改：

运行

db.rollback()

6.3 事务隔离级别详解

MySQL 支持 4 种事务隔离级别，解决并发事务中的 3 类问题：

脏读：事务 A 读取了事务 B 未提交的数据，事务 B 回滚后，A 读取的数据无效
不可重复读：事务 A 两次读取同一数据，事务 B 在两次读取之间修改并提交了数据，导致 A 两次读取结果不一致
幻读：事务 A 两次执行相同的查询，事务 B 在两次查询之间插入 / 删除了数据，导致 A 两次查询的结果集行数不一致

4 种隔离级别对比

隔离级别	脏读	不可重复读	幻读	性能	适用场景
READ UNCOMMITTED（未提交读）	允许	允许	允许	最高	数据一致性要求极低的场景（几乎不使用）
READ COMMITTED（提交读）	禁止	允许	允许	高	大多数业务场景（如电商普通订单）
REPEATABLE READ（可重复读，MySQL 默认）	禁止	禁止	允许（MySQL 通过 MVCC 解决）	中	金融、财务等数据一致性要求高的场景
SERIALIZABLE（串行化）	禁止	禁止	禁止	最低	数据一致性要求极高的场景（如银行转账）

隔离级别设置方法

运行

# 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（MySQL默认）
cursor.execute("SET TRANSACTION ISOLATION LEVEL REPEATABLE READ")

# 4. SERIALIZABLE
cursor.execute("SET TRANSACTION ISOLATION LEVEL SERIALIZABLE")

6.4 事务实战案例：转账操作

运行

import pymysql

# 连接数据库
conn = pymysql.connect(
    host="localhost",
    user="root",
    password="your_password",
    database="testdb",
    charset="utf8mb4"
)
cursor = conn.cursor()

# 关闭自动提交，开启事务
conn.autocommit = False

try:
    # 1. 从账户A扣款100元
    cursor.execute("UPDATE accounts SET balance = balance - 100 WHERE name = %s", ("Alice",))
    
    # 2. 向账户B入账100元
    cursor.execute("UPDATE accounts SET balance = balance + 100 WHERE name = %s", ("Bob",))
    
    # 提交事务
    conn.commit()
    print("转账成功")
except pymysql.MySQLError as err:
    # 发生错误，回滚事务
    print(f"转账失败: {err}")
    conn.rollback()
finally:
    # 关闭游标和连接
    cursor.close()
    conn.close()

6.5 补充：事务异常处理最佳实践

必须使用try-except-finally结构，确保异常时回滚事务
事务中避免执行耗时操作，减少锁等待时间
长事务会导致锁竞争，影响系统性能，尽量缩短事务执行时间
可通过SAVEPOINT设置事务回滚点，实现部分回滚：

运行

cursor.execute("SAVEPOINT point1")
# 部分操作
cursor.execute("ROLLBACK TO SAVEPOINT point1")

七、图片未覆盖的进阶知识点补充

7.1 异常处理与错误排查

（1）常见异常类型

pymysql.MySQLError：MySQL 相关错误的基类
pymysql.OperationalError：连接错误、权限错误、超时等
pymysql.ProgrammingError：SQL 语法错误、表不存在等
pymysql.IntegrityError：主键冲突、外键约束失败等

（2）完整异常处理模板

运行

import pymysql
from pymysql.err import MySQLError, OperationalError, ProgrammingError

try:
    # 数据库操作
    conn = pymysql.connect(...)
    cursor = conn.cursor()
    cursor.execute("SELECT * FROM users")
    conn.commit()
except OperationalError as e:
    print(f"连接错误: {e}")
    # 重连逻辑
except ProgrammingError as e:
    print(f"SQL语法错误: {e}")
except MySQLError as e:
    print(f"数据库错误: {e}")
finally:
    if 'cursor' in locals():
        cursor.close()
    if 'conn' in locals():
        conn.close()

7.2 批量操作优化

executemany()：批量插入 / 更新，性能远高于循环 execute ()
LOAD DATA INFILE：大批量数据导入（百万级以上），性能是 executemany 的 10 倍以上

运行

sql = """
LOAD DATA LOCAL INFILE 'data.csv'
INTO TABLE users
FIELDS TERMINATED BY ','
LINES TERMINATED BY '\n'
IGNORE 1 ROWS
"""
cursor.execute(sql)
conn.commit()

7.3 索引优化与慢查询

为WHERE、JOIN、ORDER BY条件列创建索引，提升查询性能
使用EXPLAIN分析 SQL 执行计划，优化慢查询

运行

cursor.execute("EXPLAIN SELECT * FROM users WHERE age > 25")
print(cursor.fetchall())

避免SELECT *，只查询需要的列，减少数据传输和内存占用

7.4 数据库连接超时与重连机制

配置connect_timeout参数，避免长时间等待
实现自动重连逻辑，应对数据库重启、网络波动等场景

运行

def get_connection():
    try:
        conn = pymysql.connect(...)
        return conn
    except OperationalError:
        # 重连3次
        for i in range(3):
            try:
                conn = pymysql.connect(...)
                return conn
            except OperationalError:
                time.sleep(1)
        raise Exception("连接数据库失败")

7.5 读写分离实现

在高并发场景下，通过读写分离提升性能：主库负责写操作，从库负责读操作

运行

# 主库（写操作）
master_conn = pymysql.connect(host="master_host", ...)
# 从库（读操作）
slave_conn = pymysql.connect(host="slave_host", ...)

# 写操作走主库
with master_conn.cursor() as cursor:
    cursor.execute("INSERT INTO users VALUES (%s, %s)", ("Eve", 22))
    master_conn.commit()

# 读操作走从库
with slave_conn.cursor() as cursor:
    cursor.execute("SELECT * FROM users")
    results = cursor.fetchall()

八、完整实战案例：用户管理系统

运行

import pymysql
from dbutils.pooled_db import PooledDB

# 1. 初始化连接池
dbconfig = {
    "host": "localhost",
    "user": "root",
    "password": "your_password",
    "database": "testdb",
    "port": 3306,
    "charset": "utf8mb4"
}
pool = PooledDB(
    creator=pymysql,
    maxconnections=5,
    mincached=2,
    ping=1,
    **dbconfig
)

# 2. 用户CRUD操作
class UserManager:
    def __init__(self):
        self.conn = pool.connection()
        self.cursor = self.conn.cursor()
    
    def add_user(self, name, age):
        """添加用户"""
        try:
            sql = "INSERT INTO users (name, age) VALUES (%s, %s)"
            self.cursor.execute(sql, (name, age))
            self.conn.commit()
            return True
        except pymysql.MySQLError as e:
            print(f"添加用户失败: {e}")
            self.conn.rollback()
            return False
    
    def get_user_by_name(self, name):
        """根据姓名查询用户"""
        sql = "SELECT * FROM users WHERE name = %s"
        self.cursor.execute(sql, (name,))
        return self.cursor.fetchone()
    
    def update_user_age(self, name, new_age):
        """更新用户年龄"""
        try:
            sql = "UPDATE users SET age = %s WHERE name = %s"
            self.cursor.execute(sql, (new_age, name))
            self.conn.commit()
            return True
        except pymysql.MySQLError as e:
            print(f"更新用户失败: {e}")
            self.conn.rollback()
            return False
    
    def delete_user(self, name):
        """删除用户"""
        try:
            sql = "DELETE FROM users WHERE name = %s"
            self.cursor.execute(sql, (name,))
            self.conn.commit()
            return True
        except pymysql.MySQLError as e:
            print(f"删除用户失败: {e}")
            self.conn.rollback()
            return False
    
    def close(self):
        """关闭连接（归还到连接池）"""
        self.cursor.close()
        self.conn.close()

# 3. 测试
if __name__ == "__main__":
    manager = UserManager()
    # 添加用户
    manager.add_user("Eve", 22)
    # 查询用户
    print(manager.get_user_by_name("Eve"))
    # 更新年龄
    manager.update_user_age("Eve", 23)
    # 删除用户
    manager.delete_user("Eve")
    # 关闭连接
    manager.close()

九、总结

本文完整覆盖了 Python 操作 MySQL 的全流程技术：从基础的连接库安装、数据库连接与 CRUD 操作，到进阶的连接池优化、事务管理与隔离级别，同时补充了异常处理、SQL 注入防护、批量操作、读写分离等实战知识点。

核心要点回顾：

基础操作：使用 PyMySQL 建立连接，通过游标执行 SQL，%s占位符防注入
性能优化：使用 DBUtils 连接池，减少连接开销，提升高并发性能
数据一致性：通过事务管理保障数据安全，根据业务选择合适的隔离级别
进阶技巧：异常处理、批量操作、索引优化、读写分离等，解决实际开发中的各类问题

掌握这些技能，不仅能满足运维自动化、后端开发中的数据库操作需求，还能解决高并发场景下的性能瓶颈与数据一致性问题，是运维工程师、后端开发的必备核心技能。

参考资料：

PyMySQL 官方文档：https://pymysql.readthedocs.io/
DBUtils 官方文档：https://webwareforpython.github.io/DBUtils/
MySQL 事务隔离级别官方文档：https://dev.mysql.com/doc/refman/8.0/en/innodb-transaction-isolation-levels.html
Python MySQL 连接库对比：https://realpython.com/python-mysql/

以上为个人经验，希望能给大家一个参考，也希望大家多多支持脚本之家。

您可能感兴趣的文章:

Pandas分组函数groupby的用法详解
在数据分析时,经常需要将数据分成不同的群组,pandas中的groupby()函数可以完美地完成各种分组操作,本文就来介绍一下Pandas分组函数groupby的用法,感兴趣的可以了解一下
2024-01-01
Python检测数据类型的方法总结
在本篇文章里小编给大家整理了关于Python检测数据类型的方法和相关实例代码，需要的朋友们跟着学习下。
2019-05-05
Python之plt.bar绘制柱状图参数解读
这篇文章主要介绍了Python之plt.bar绘制柱状图参数,具有很好的参考价值,希望对大家有所帮助,如有错误或未考虑完全的地方,望不吝赐教
2023-09-09
JMETER如何随机请求参数
作者分享了在业务场景中如何优化接口查询的实践经验,主要面临的问题是需要随机获取上游查询接口的结果,提出了两种方案,方案一是反复查询并随机获取查询结果；方案二是查询一次,然后随机从查询结果中获取,通过实测比较,方案二的性能提升较大
2024-10-10
python对列表中任意两个数进行操作的实现
本文主要介绍了在Python中实现列表中整型元素和数组元素两两相乘或两两相与的操作,具有一定的参考价值,感兴趣的可以了解一下
2025-01-01
python递归函数求n的阶乘,优缺点及递归次数设置方式
这篇文章主要介绍了python递归函数求n的阶乘,优缺点及递归次数设置方式，具有很好的参考价值，希望对大家有所帮助。一起跟随小编过来看看吧
2020-04-04
Python 中 Selenium 的 send_keys() 函数用法小结
send_keys() 是将数字、文本和符号等键盘输入发送到应用程序的文本框的过程， send_keys() 是 WebDriver 的一部分，每个键盘输入都会发送到此元素，这篇文章主要介绍了Python 中 Selenium 的 send_keys() 函数,需要的朋友可以参考下
2023-11-11
Python异步爬虫requests和aiohttp中代理IP的使用
本文主要介绍了Python异步爬虫requests和aiohttp中代理IP的使用，文中通过示例代码介绍的非常详细，具有一定的参考价值，感兴趣的小伙伴们可以参考一下
2022-03-03
python实现简单聊天应用 python群聊和点对点均实现
这篇文章主要为大家详细介绍了python实现简单聊天应用，python群聊和点对点均实现，具有一定的参考价值，感兴趣的小伙伴们可以参考一下
2017-09-09
Python : turtle色彩控制实例详解
今天小编就为大家分享一篇Python : turtle色彩控制实例详解，具有很好的参考价值，希望对大家有所帮助。一起跟随小编过来看看吧
2020-01-01