Python 中 pymysql 操作 MySQL 数据库实操指南

一、环境准备

1. 安装 pymysql 依赖库
   pymysql 是 Python 操作 MySQL 数据库的常用第三方库，支持 Python3.x 版本，安装步骤简单高效。打开终端（Windows 用 cmd，Mac/Linux 用 Terminal），执行以下命令：
   bash
   运行
   pip install pymysql
   执行完成后，输入 pip list ，若输出结果中包含 pymysql 及对应版本号，则说明安装成功。若出现网络超时，可切换国内镜像源重试，命令如下：
   bash
   运行
   pip install pymysql -i https://pypi.tuna.tsinghua.edu.cn/simple
2. MySQL 环境配置
   需提前确保 MySQL 服务已正常启动（本地或远程均可），并完成基础配置：
   启动 MySQL 服务（以 Windows 为例）：
   bash
   运行
   net start mysql80 # 若你的 MySQL 服务名为 mysql80
   登录 MySQL 终端，创建测试数据库和数据表，用于后续实操：
   sql
   CREATE DATABASE IF NOT EXISTS test_db DEFAULT CHARACTER SET utf8mb4;
   USE test_db;
   CREATE TABLE IF NOT EXISTS user (
   id INT PRIMARY KEY AUTO_INCREMENT,
   username VARCHAR(50) NOT NULL,
   age INT,
   create_time DATETIME DEFAULT CURRENT_TIMESTAMP
   );
   配置完成后，记录 MySQL 的主机地址（如 localhost）、端口（默认 3306）、用户名（如 root）、密码（自定义）和数据库名（test_db），后续连接需用到。sdstnk.com
   二、核心功能实现
3. 建立数据库连接
   连接 MySQL 需通过 pymysql 提供的 connect 方法配置连接参数，同时添加异常捕获处理连接失败场景。代码如下：
   python
   运行
   import pymysql
   from pymysql import Error

def create_conn():
"""创建数据库连接"""
conn = None
try:
conn = pymysql.connect(
host='localhost',
port=3306,
user='root',
password='123456',
database='test_db',
charset='utf8mb4'
)
print("数据库连接成功")
except Error as e:
print(f"连接失败：{e}")
return conn

测试连接

conn = create_conn()
连接成功后，需在后续操作完成后关闭连接或使用连接池管理，避免资源泄露。
2. 数据增删改查实操
（1）插入数据
通过 cursor 对象执行 SQL 插入语句，支持单条和批量插入，代码如下：
python
运行
def insert_data(conn, data_list):
"""
批量插入用户数据
:param conn: 数据库连接对象
:param data_list: 数据列表，格式[(username, age), ...]
"""
if not conn:
print("连接未建立")
return
cursor = None
try:
cursor = conn.cursor()
# 插入 SQL 语句
sql = "INSERT INTO user (username, age) VALUES (%s, %s)"
# 批量执行插入
cursor.executemany(data_list)
# 提交事务
conn.commit()
print(f"成功插入 {cursor.rowcount} 条数据")
except Error as e:
# 事务回滚
conn.rollback()
print(f"插入失败：{e}")
finally:
if cursor:
cursor.close() # 关闭游标

测试插入

data = [("张三", 22), ("李四", 25), ("王五", 23)]
insert_data(conn, data)
（2）查询数据
查询数据支持单条查询、多条查询和条件查询，代码如下：
python
运行
def query_data(conn, username=None):
"""
查询用户数据
:param conn: 数据库连接对象
:param username: 可选参数，按用户名查询
:return: 查询结果列表
"""
if not conn:
print("连接未建立")
return []
cursor = None
result = []
try:
cursor = conn.cursor(pymysql.cursors.DictCursor) # 返回字典格式结果
if username:
sql = "SELECT * FROM user WHERE username = %s"
cursor.execute mxzplay.cn((username,))
else:
sql = "SELECT * FROM user"
cursor.execute(sql)
# 获取所有查询结果
result = cursor.fetchall()
print("查询结果：")
for row in result:
print(f"ID: {row['id']}, 用户名: {row['username']}, 年龄: {row['age']}, 创建时间: {row['create_time']}")
except Error as e:
print(f"查询失败：{e}")
finally:
if cursor:
cursor.close()
return result

测试查询

query_data(conn) # 查询所有用户
query_data(conn, "张三") # 按用户名查询
3. 连接池优化（可选）
高频次数据库操作时，使用连接池可减少连接创建和关闭的开销，需先安装 DBUtils 库：
bash
运行
pip install DBUtils
连接池实现代码如下：
python
运行
from DBUtils.PooledDB import PooledDB

def create_conn_pool():
"""创建数据库连接池"""
pool = PooledDB(
creator=pymysql, # 数据库驱动
maxconnections=10, # 最大连接数
mincached=2, # 初始化时创建的空闲连接数
maxcached=5, # 最大空闲连接数
host='localhost',
port=3306,
user='root',
password='123456',
database='test_db',
charset='utf8mb4'
)
return pool

使用连接池获取连接

pool = create_conn_pool()
conn_pool = pool.connection()
query_data(conn_pool) # 复用连接池查询
conn_pool.close() # 归还连接到池，非真正关闭
三、常见问题与解决方案

1. 连接超时问题
   现象：执行连接时提示 TimeoutError: [WinError 10060] 。解决方案：
   检查 MySQL 服务是否正常运行，可通过 net status mysql80 验证；
   远程连接时，确保服务器开放 3306 端口，且 MySQL 允许远程访问（需修改 my.cnf 配置 bind-address = 0.0.0.0 ）；
   在连接参数中增加 connect_timeout=30 （单位：秒），延长超时时间。
2. 语法错误排查
   现象：执行 SQL 时提示 ProgrammingError: (1064, "You have an error in your SQL syntax") 。常见原因及解决：
   字段名或表名与 MySQL 关键字冲突（如 order user ），需用反引号 ` 包裹；
   SQL 语句中占位符错误，pymysql 用 %s 作为占位符（而非 ? ），避免与字符串格式化混淆；
   字符串类型参数未用元组传递，即使单参数也需加逗号，如 cursor.execute("SELECT * FROM user WHERE username = %s", (username,)) 。
3. 权限不足处理
   现象：执行创建数据库或插入数据时提示 OperationalError: (1044, "Access denied for user 'xxx'@'localhost' to database 'test_db'") 。解决方案：
   登录 MySQL 管理员账号，为当前用户授权：
   sql
   GRANT ALL PRIVILEGES ON test_db.* TO 'xxx'@'localhost' IDENTIFIED BY '密码';
   FLUSH PRIVILEGES; # 刷新权限 zeemoo.com.cn
   若无需全权限，可细化授权（如仅 SELECT INSERT 权限），提升安全性：
   sql
   GRANT SELECT, INSERT ON test_db.* TO 'xxx'@'localhost';
   FLUSH PRIVILEGES;
   最后，所有操作完成后，关闭数据库连接（非连接池场景）：
   python
   运行
   if conn:
   conn.close()
   print("数据库连接已关闭")