Python使用PyMySQL连接MySQL数据库完整代码示例
在 Python 开发中，与 MySQL 数据库交互最轻量且通用的方式之一是使用 PyMySQL。它是一个基于纯 Python 实现的驱动程序，完全兼容 Python DB-API 2.0 规范，不需要在操作系统中预先安装底层的 C 语言客户端开发包。本文通过一套系统、规范且可直接运行的代码示例，带你全面掌握从环境准备、连接参数调优、CRUD（增删改查）操作、事务回滚到连接上下文管理的全流程。
一、PyMySQL 通信架构与运作原理
为了理解代码中各对象的协作方式，我们先梳理 Python 应用程序与 MySQL 数据库服务之间的交互链路。PyMySQL 作为客户端驱动，承担了将 Python 对象序列化为 MySQL 通信报文、通过 TCP Socket 发送给服务端，并将服务端返回的原始二进制数据解码为 Python 常用数据类型（如字典、元组、字符串、日期等）的任务。
二、环境依赖安装与测试数据表准备
在开始编写代码前，首先需要在 Python 运行环境中安装 PyMySQL 依赖，并在 MySQL 数据库中准备测试用的数据库和数据表。
1. 安装 PyMySQL
打开终端或命令行，执行以下 pip 命令：
pip install pymysql如果在 Django 或某些依赖 mysqlclient 的框架中使用 PyMySQL，可通过在项目入口处加入 pymysql.install_as_MySQLdb() 实现无缝伪装替换。
2. 准备测试数据表
在 MySQL 中创建一个名为 test_db 的数据库以及一张用于演示的学生信息表 students：
CREATE DATABASE IF NOT EXISTS test_db DEFAULT CHARACTER SET utf8mb4 COLLATE utf8mb4_unicode_ci;
USE test_db;
CREATE TABLE IF NOT EXISTS students (
id INT AUTO_INCREMENT PRIMARY KEY,
name VARCHAR(50) NOT NULL COMMENT '姓名',
age INT NOT NULL COMMENT '年龄',
email VARCHAR(100) DEFAULT NULL COMMENT '邮箱',
created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP COMMENT '创建时间'
) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4;三、建立数据库连接与核心参数解析
连接 MySQL 数据库的入口函数是 pymysql.connect()。该函数接受一系列关键配置项：
| 参数名 | 类型 | 示例值 | 说明与最佳实践 |
|---|---|---|---|
host |
str | '127.0.0.1' | MySQL 服务器主机地址。推荐使用 IP 避免本机 localhost 域名解析开销。 |
port |
int | 3306 | MySQL 监听端口，必须为正整数。 |
user |
str | 'root' | 连接数据库所用的用户名。 |
password |
str | 'your_password' | 数据库用户密码。 |
database |
str | 'test_db' | 要操作的默认数据库名称（也可写作 db）。 |
charset |
str | 'utf8mb4' | 字符编码。强烈建议使用 utf8mb4，完整支持 Emoji 及 4 字节特殊符号。 |
cursorclass |
Cursor | pymysql.cursors.DictCursor |
游标类型。默认为元组游标；指定为 DictCursor 后，查询结果以字典形式返回。 |
autocommit |
bool | False | 是否自动提交。生产环境下建议保持 False，由程序显式控制事务。 |
四、操作步骤流程图
使用 PyMySQL 进行数据库操作的标准流程分为五个核心环节：从建立连接、获取游标，到执行 SQL、处理结果与事务确认，最后安全关闭资源。
五、完整的 CRUD 数据库操作代码示例
下面是一份完整的 Python 脚本，涵盖了插入（Insert）、批量插入、单条查询、全量查询、数据更新（Update）及数据删除（Delete）。可以直接保存并执行。
import pymysql
from pymysql.cursors import DictCursor
def get_connection():
"""创建并返回数据库连接对象"""
return pymysql.connect(
host='127.0.0.1',
port=3306,
user='root',
password='your_password', # 替换为你的实际数据库密码
database='test_db',
charset='utf8mb4',
cursorclass=DictCursor, # 查询结果映射为 Python 字典
autocommit=False # 关闭自动提交，显式控制事务
)
def main():
conn = get_connection()
try:
with conn.cursor() as cursor:
# 1. 单条插入数据 (INSERT)
print("--- 1. 执行单条插入 ---")
insert_sql = """
INSERT INTO students (name, age, email)
VALUES (%s, %s, %s);
"""
cursor.execute(insert_sql, ("张三", 20, "zhangsan@example.com"))
# 获取刚刚自增生成的主键 ID
new_id = cursor.lastrowid
print(f"成功插入数据，新增记录 ID: {new_id}")
# 2. 批量插入数据 (executemany)
print("\n--- 2. 执行批量插入 ---")
batch_data = [
("李四", 22, "lisi@example.com"),
("王五", 21, "wangwu@example.com"),
("赵六", 23, "zhaoliu@example.com")
]
affected_rows = cursor.executemany(insert_sql, batch_data)
print(f"批量插入完成，受影响行数: {affected_rows}")
# 写入操作完成后显式提交事务
conn.commit()
print("事务已提交！")
# 3. 单条查询 (fetchone)
print("\n--- 3. 执行单条查询 ---")
select_one_sql = "SELECT id, name, age, email, created_at FROM students WHERE id = %s;"
cursor.execute(select_one_sql, (new_id,))
student = cursor.fetchone()
print("单条查询结果:", student)
# 4. 全量查询 (fetchall)
print("\n--- 4. 执行全量查询 ---")
select_all_sql = "SELECT id, name, age, email FROM students WHERE age >= %s ORDER BY id ASC;"
cursor.execute(select_all_sql, (20,))
all_students = cursor.fetchall()
print(f"共查询到 {len(all_students)} 条记录:")
for s in all_students:
print(f" ID: {s['id']} | 姓名: {s['name']} | 年龄: {s['age']} | 邮箱: {s['email']}")
# 5. 更新数据 (UPDATE)
print("\n--- 5. 执行更新操作 ---")
update_sql = "UPDATE students SET age = %s, email = %s WHERE id = %s;"
update_rows = cursor.execute(update_sql, (25, "zhangsan_new@example.com", new_id))
print(f"更新完成，受影响行数: {update_rows}")
conn.commit()
# 6. 删除数据 (DELETE)
print("\n--- 6. 执行删除操作 ---")
delete_sql = "DELETE FROM students WHERE name = %s;"
del_rows = cursor.execute(delete_sql, ("赵六",))
print(f"删除完成，受影响行数: {del_rows}")
conn.commit()
except pymysql.MySQLError as err:
# 捕获异常后必须回滚未提交的脏数据
conn.rollback()
print(f"执行数据库操作失败，已回滚事务！错误信息: {err}")
finally:
# 确保连接最终被安全关闭
conn.close()
print("\n数据库连接已安全关闭。")
if __name__ == '__main__':
main()六、防范 SQL 注入与参数化查询规范
在使用 PyMySQL 时，最常见且危险的安全漏洞是采用 Python 的字符串格式化（如 f-string 或 % 操作符）手动拼接 SQL。这种做法将用户输入当做 SQL 语法的一部分解析，极易导致 SQL 注入攻击。
# 漏洞代码：若 user_input 为 "' OR '1'='1"，整张表将被误删或泄露
cursor.execute(f"SELECT * FROM students WHERE name = '{user_input}'")
正确的安全规范：必须在 SQL 语句中使用 %s 占位符，并将参数以元组或列表形式作为第二个参数传递给 execute()。注意：无论数据库字段是整数、浮点数还是字符串，在 PyMySQL 的 SQL 模板中一律统一使用 %s 占位，且占位符外绝对不能加引号：
# 安全示例：PyMySQL 会自动对参数进行严格转义与类型安全映射
safe_sql = "SELECT * FROM students WHERE name = %s AND age > %s;"
cursor.execute(safe_sql, (user_name, min_age))七、事务控制与异常回滚机制
在数据库写操作（INSERT、UPDATE、DELETE）中，数据一致性与原子性至关重要。PyMySQL 默认不会在每次 execute 后自动提交，必须调用 conn.commit() 才能将修改真正持久化到磁盘中。如果中途发生异常未做提交，连接断开后所有修改将全部丢失。
为了保障多步写操作的原子性，必须采用 try...except...finally 控制结构，当任一环节出错时立即触发 conn.rollback() 回滚。
八、生产环境健壮实践：上下文管理器封装
在大型应用中，手动调用 close() 容易因疏忽导致连接或游标泄漏。借助 Python 的 contextlib 模块，我们可以将数据库连接封装为上下文管理器（Context Manager），实现出作用域时自动提交事务、发生异常时自动回滚、退出代码块时自动关闭连接的高内聚模式：
import pymysql
from pymysql.cursors import DictCursor
from contextlib import contextmanager
DB_CONFIG = {
'host': '127.0.0.1',
'port': 3306,
'user': 'root',
'password': 'your_password',
'database': 'test_db',
'charset': 'utf8mb4',
'cursorclass': DictCursor,
'connect_timeout': 10
}
@contextmanager
def db_session():
"""生产级数据库连接上下文管理器：自动管理事务与连接生命周期"""
conn = pymysql.connect(**DB_CONFIG)
try:
yield conn
conn.commit() # 代码块正常执行结束则自动提交
except Exception as exc:
conn.rollback() # 捕获任何异常自动回滚
print(f"事务执行失败，已自动回滚: {exc}")
raise
finally:
conn.close() # 无论成功失败均自动关闭连接
# 使用方式示例：
def query_user(user_id):
with db_session() as conn:
with conn.cursor() as cursor:
cursor.execute("SELECT * FROM students WHERE id = %s;", (user_id,))
return cursor.fetchone()九、核心配置与避坑要点总结
为便于日常开发查阅，我们将 PyMySQL 最容易踩坑的四个维度整理如下：
常见报错与排查建议
OperationalError: (1045, "Access denied for user...")：账号或密码错误，或者该用户未授权从当前主机（如'root'@'localhost'与'root'@'%'）登录。OperationalError: (2003, "Can't connect to MySQL server...")：MySQL 服务未启动、IP 地址错误、端口被防火墙拦截，或者 MySQL 配置中bind-address被限制。InternalError: (1366, "Incorrect string value...")：插入包含生僻字或 Emoji 文本时抛出，通常是因为连接时未指定charset='utf8mb4'或目标表字段编码为utf8/latin1。
免责声明：本文内容仅供参考，不代表本站观点。转载或引用请注明出处。
