【第十一章】Python 调用 MySQL 全面指南：从基础到实践 - 实践

Python 调用 MySQL 全面指南：从基础到实践

在当今数据驱动的时代，Python 凭借其简洁的语法、丰富的生态库，成为数据处理、Web 开发、自动化脚本编写的首选语言之一；而 MySQL 作为开源关系型数据库的代表，以其高效、稳定、易扩展的特性，广泛应用于各类系统的数据存储场景。两者的结合，是实现 “数据读取 - 处理 - 存储” 闭环的核心环节 —— 无论是电商系统的订单数据管理、数据分析平台的批量数据入库，还是自动化脚本的日志存储，都离不开 Python 对 MySQL 的调用。

前言

对于初学者而言，从环境搭建到异常处理，从单条数据操作到高并发场景下的连接池优化，往往存在诸多疑问。本文旨在通过 “理论 + 实例 + 执行结果” 的形式，覆盖 Python 操作 MySQL 的全流程知识点，既解决 “怎么用” 的问题，也解释 “为什么这么用”，帮助读者不仅能完成基础交互，更能规避常见坑点，写出高效、安全的数据库操作代码。

在数据处理与开发中，Python 常需与 MySQL 数据库交互以实现数据存储、查询等功能。本文将覆盖环境搭建、连接配置、SQL 执行、结果处理、事务管理及异常处理等核心知识点，通过代码示例与执行结果，助你快速掌握 Python 操作 MySQL 的完整流程。

一、环境准备：安装必备库

Python 操作 MySQL 需依赖第三方库，常用库有mysql-connector-python（Oracle 官方）和pymysql（Python 社区维护），两者用法类似，本文以mysql-connector-python为例。

1.1 安装库

通过pip命令安装，执行以下命令：

1、pip install mysql-connector-python
2、若使用pymysql，执行：pip install pymysql

1.2 验证安装

进入 Python 交互环境，执行导入命令，无报错则安装成功：

import mysql.connector
print("安装成功") # 执行结果：安装成功

二、核心操作：Python 与 MySQL 交互全流程

2.1 连接 MySQL 数据库

连接数据库需指定主机地址、用户名、密码、数据库名等参数，通过connect()方法创建连接对象，后续操作均基于该对象。

示例代码：

import mysql.connector
from mysql.connector import Error
try:
# 1. 创建数据库连接
connection = mysql.connector.connect(
host='localhost', # 主机地址（本地为localhost）
database='test_db', # 要连接的数据库名（需提前创建）
user='root', # MySQL用户名
password='Admin@123' # MySQL密码
)
# 2. 验证连接
if connection.is_connected():
db_info = connection.server_info
print(f"连接成功！MySQL服务器版本：{db_info
}") # 执行结果：连接成功！MySQL服务器版本：8.0.32
# 获取当前数据库游标（用于执行SQL）
cursor = connection.cursor()
# 查询当前使用的数据库
cursor.execute("select database();")
current_db = cursor.fetchone()
print(f"当前连接的数据库：{current_db[0]
}") # 执行结果：当前连接的数据库：test_db
except Error as e:
print(f"连接失败！错误信息：{e
}") # 若参数错误，执行结果例：连接失败！错误信息：1045 (28000): Access denied for user 'root'@'localhost' (using password: YES)
finally:
# 3. 关闭连接（避免资源泄漏）
if connection.is_connected():
cursor.close()
connection.close()
print("数据库连接已关闭") # 执行结果：数据库连接已关闭

2.2 执行 SQL 操作（增删改查）

通过游标对象cursor的execute()方法执行 SQL 语句，不同操作的结果处理方式不同：查询（SELECT）需获取结果，增删改（INSERT/DELETE/UPDATE）需提交事务。

2.2.1 创建数据表（CREATE TABLE）

# 承接上文的connection和cursor对象
create_table_sql = """
CREATE TABLE IF NOT EXISTS student (
id INT AUTO_INCREMENT PRIMARY KEY, # 自增主键
name VARCHAR(50) NOT NULL, # 姓名（非空）
age INT, # 年龄
score FLOAT # 分数
) ENGINE=InnoDB DEFAULT CHARSET=utf8;
"""
cursor.execute(create_table_sql)
print("数据表student创建成功") # 执行结果：数据表student创建成功

2.2.2 插入数据（INSERT）

插入单条或多条数据，需通过connection.commit()提交事务，否则数据不会写入数据库。

示例 1：插入单条数据

insert_sql = "INSERT INTO student (name, age, score) VALUES (%s, %s, %s);"
data = ("张三", 18, 95.5) # 与SQL中的%s对应（注意：不是%s字符串拼接，避免SQL注入）
cursor.execute(insert_sql, data)
connection.commit() # 提交事务
print(f"插入成功，新增数据ID：{cursor.lastrowid
}") # 执行结果：插入成功，新增数据ID：1

示例 2：插入多条数据

insert_many_sql = "INSERT INTO student (name, age, score) VALUES (%s, %s, %s);"
data_list = [
("李四", 19, 88.0),
("王五", 17, 92.5),
("赵六", 18, 79.0)
]
cursor.executemany(insert_many_sql, data_list) # executemany批量执行
connection.commit()
print(f"批量插入成功，共插入{cursor.rowcount
}条数据") # 执行结果：批量插入成功，共插入3条数据

2.2.3 查询数据（SELECT）

查询结果通过cursor.fetchone()（获取一条）、cursor.fetchall()（获取所有）、cursor.fetchmany(n)（获取 n 条）获取，结果以元组或元组列表形式返回。

示例：查询所有学生数据

select_sql = "SELECT id, name, age, score FROM student WHERE age < 20;"
cursor.execute(select_sql)
# 获取所有查询结果
results = cursor.fetchall()
print(f"查询到{cursor.rowcount
}条数据：") # 执行结果：查询到4条数据：
# 遍历结果
for row in results:
student_id = row[0]
name = row[1]
age = row[2]
score = row[3]
print(f"ID：{student_id
}，姓名：{name
}，年龄：{age
}，分数：{score
}")

# 执行结果：
# ID：1，姓名：张三，年龄：18，分数：95.5
# ID：2，姓名：李四，年龄：19，分数：88.0
# ID：3，姓名：王五，年龄：17，分数：92.5
# ID：4，姓名：赵六，年龄：18，分数：79.0

2.2.4 更新数据（UPDATE）

update_sql = "UPDATE student SET score = %s WHERE name = %s;"
update_data = (90.0, "赵六") # 将赵六的分数改为90.0
cursor.execute(update_sql, update_data)
connection.commit()
print(f"更新成功，影响{cursor.rowcount
}条数据") # 执行结果：更新成功，影响1条数据

2.2.5 删除数据（DELETE）

delete_sql = "DELETE FROM student WHERE age = %s;"
delete_data = (17,) # 注意：单参数需加逗号，避免被识别为普通元组
cursor.execute(delete_sql, delete_data)
connection.commit()
print(f"删除成功，影响{cursor.rowcount
}条数据") # 执行结果：删除成功，影响1条数据

2.3 事务管理

MySQL 默认开启事务（InnoDB 引擎），需通过commit()提交事务，或通过rollback()回滚事务（出现错误时），确保数据一致性。

示例：事务回滚

try:
# 执行两个操作，作为一个事务
cursor.execute("INSERT INTO student (name, age, score) VALUES ('孙七', 20, 85.0);")
# 模拟错误（如分数超过100，实际业务中可能是逻辑错误）
cursor.execute("INSERT INTO student (name, age, score) VALUES ('周八', 21, 105.0);")
connection.commit()
print("事务提交成功")
except Error as e:
# 出现错误，回滚事务（两条插入均不生效）
connection.rollback()
print(f"事务回滚！错误信息：{e
}") # 执行结果：事务回滚！错误信息：...（若无约束，仅提示执行成功，可自定义业务逻辑触发回滚）

2.4 异常处理最佳实践

实际开发中，需捕获常见异常（如连接失败、SQL 语法错误、数据约束错误），并给出清晰提示，避免程序崩溃。

示例：完整异常处理模板

import mysql.connector
from mysql.connector import Error
def operate_mysql():
connection = None
cursor = None
try:
# 1. 连接数据库
connection = mysql.connector.connect(
host='localhost',
database='test_db',
user='root',
password='123456'
)
if connection.is_connected():
cursor = connection.cursor()
# 2. 执行SQL（示例：查询）
cursor.execute("SELECT name, score FROM student;")
results = cursor.fetchall()
print("学生列表：", results)
except Error as e:
# 分类处理异常
if "Access denied" in str(e):
print("错误：用户名或密码错误")
elif "Unknown database" in str(e):
print("错误：数据库不存在")
elif "You have an error in your SQL syntax" in str(e):
print(f"错误：SQL语法错误：{e
}")
else:
print(f"未知错误：{e
}")
finally:
# 3. 关闭资源（无论是否报错，均需关闭）
if cursor:
cursor.close()
if connection and connection.is_connected():
connection.close()
print("连接已关闭")
# 调用函数
operate_mysql()
# 执行结果（若正常）：
# 学生列表： [('张三', 95.5), ('李四', 88.0), ('赵六', 90.0), ('孙七', 85.0)]
# 连接已关闭

三、进阶技巧：配置文件与连接池

3.1 从配置文件读取连接参数

将数据库参数（如密码）写入配置文件（如config.ini），避免硬编码，提高安全性。

步骤 1：创建 config.ini

[mysql]
host = localhost
database = test_db
user = root
password = Admin@123
port = 3306

步骤 2：读取配置并连接

import configparser # 用于读取ini文件
# 读取配置
config = configparser.ConfigParser()
config.read('config.ini')
mysql_config = config['mysql']
# 连接数据库
connection = mysql.connector.connect(
host=mysql_config['host'],
database=mysql_config['database'],
user=mysql_config['user'],
password=mysql_config['password'],
port=int(mysql_config['port'])
)
print("从配置文件连接成功") # 执行结果：从配置文件连接成功

3.2 使用连接池（提高性能）

频繁创建 / 关闭连接会消耗资源，通过连接池管理连接，可复用连接，提升高并发场景下的性能。

示例：创建连接池

from mysql.connector import pooling
# 创建连接池
connection_pool = pooling.MySQLConnectionPool(
pool_name="my_pool", # 连接池名称
pool_size=5, # 最大连接数（根据需求调整）
pool_reset_session=True, # 重置会话（避免前一个连接的影响）
host='localhost',
database='test_db',
user='root',
password='123456'
)
# 从连接池获取连接
connection1 = connection_pool.get_connection()
connection2 = connection_pool.get_connection()
print(f"连接池当前可用连接数：{connection_pool.pool_size - connection_pool.used_connections
}") # 执行结果：连接池当前可用连接数：3
# 关闭连接（实际是归还到连接池，而非销毁）
connection1.close()
print(f"归还连接后，可用连接数：{connection_pool.pool_size - connection_pool.used_connections
}") # 执行结果：归还连接后，可用连接数：4

四、常见问题与解决方案

问题现象	可能原因	解决方案
连接失败：Access denied	用户名 / 密码错误	检查user和password参数，确保与 MySQL 配置一致
连接失败：Unknown database	数据库不存在	先通过 MySQL 客户端创建数据库（CREATE DATABASE test_db;）
执行 SQL 报错：You have an error in your SQL syntax	SQL 语句语法错误	检查 SQL 关键字、括号、分号，确保与 MySQL 语法一致
插入数据后查询不到	未执行connection.commit()	增删改操作后必须调用commit()提交事务
批量插入效率低	未使用executemany()	优先使用executemany()批量执行，而非循环execute()