在Python后端开发、数据分析等场景中，MySQL是最常用的关系型数据库之一，而pymysql则是Python连接MySQL的主流第三方库。本文将从环境准备、数据库连接、基础数据读写、事务处理到异常处理，全方位讲解如何用pymysql实现Python与MySQL的交互，内容贴合实际开发场景，新手也能快速上手。

一、前置准备：环境与依赖安装

1. 检查MySQL环境

确保本地或服务器已安装MySQL数据库，且能正常启动。可通过以下命令验证（Windows/Linux通用，需配置MySQL环境变量）：

mysql -u root -p

输入密码后能进入MySQL终端，说明环境正常。

2. 安装pymysql库

pymysql不是Python标准库，需通过pip安装，命令如下：

# 基础安装

pip install pymysql

# 指定版本（推荐稳定版）

pip install pymysql==1.1.0

安装完成后，在Python脚本中导入pymysql，无报错即安装成功：

import pymysql

二、核心步骤：Python连接MySQL数据库

1. 基础连接语法

pymysql通过connect()方法创建数据库连接对象，核心参数说明：

参数名

说明

示例

host

MySQL服务器地址

localhost/192.168.1.100

port

端口号，默认3306

3306

user

数据库用户名

root

password

数据库密码

123456

database

要连接的数据库名

test_db

charset

字符编码，推荐utf8mb4

utf8mb4

2. 最简连接示例

import pymysql

# 1. 创建数据库连接

conn = pymysql.connect(

host='localhost',

port=3306,

user='root',

password='你的MySQL密码',

database='test_db',

charset='utf8mb4'

)

# 2. 验证连接（可选）

try:

# 获取游标对象（用于执行SQL）

cursor = conn.cursor()

print("数据库连接成功！")

finally:

# 3. 关闭游标和连接（避免资源泄露）

cursor.close()

conn.close()

3. 连接池优化（生产环境推荐）

频繁创建/关闭连接会消耗资源，生产环境建议使用连接池。这里推荐pymysqlpool（需先安装pip install pymysqlpool）：

from pymysqlpool import ConnectionPool

# 配置连接池

pool = ConnectionPool(

host='localhost',

port=3306,

user='root',

password='你的MySQL密码',

database='test_db',

charset='utf8mb4',

max_size=10, # 最大连接数

min_size=2 # 最小连接数

)

# 获取连接

conn = pool.get_connection()

cursor = conn.cursor()

# 用完归还连接

cursor.close()

conn.close()

三、数据读写：增删改查（CRUD）实操

1. 数据查询（SELECT）

查询是最常用的操作，需注意游标执行SQL后，通过fetchone()/fetchall()/fetchmany()获取结果。

示例1：查询单条数据

import pymysql

def query_single_data():

conn = None

cursor = None

try:

# 创建连接

conn = pymysql.connect(

host='localhost',

port=3306,

user='root',

password='你的密码',

database='test_db',

charset='utf8mb4'

)

# 创建游标（指定返回字典格式，更易读）

cursor = conn.cursor(cursor=pymysql.cursors.DictCursor)

# 执行查询SQL

sql = "SELECT id, name, age FROM user WHERE id = %s"

cursor.execute(sql, (1,)) # 参数化查询，避免SQL注入

# 获取结果

result = cursor.fetchone()

print("查询结果：", result)

except pymysql.MySQLError as e:

print("查询失败：", e)

finally:

# 关闭资源

if cursor:

cursor.close()

if conn:

conn.close()

if __name__ == '__main__':

query_single_data()

示例2：查询多条数据

def query_multi_data():

conn = None

cursor = None

try:

conn = pymysql.connect(

host='localhost',

port=3306,

user='root',

password='你的密码',

database='test_db',

charset='utf8mb4'

)

cursor = conn.cursor(cursor=pymysql.cursors.DictCursor)

# 查询年龄大于18的用户

sql = "SELECT id, name, age FROM user WHERE age > %s"

cursor.execute(sql, (18,))

# 获取所有结果

results = cursor.fetchall()

print("查询结果总数：", cursor.rowcount)

for row in results:

print(f"ID: {row['id']}, 姓名: {row['name']}, 年龄: {row['age']}")

except pymysql.MySQLError as e:

print("查询失败：", e)

finally:

if cursor:

cursor.close()

if conn:

conn.close()

2. 数据插入（INSERT）

插入数据后需调用conn.commit()提交事务，否则数据不会写入数据库。

def insert_data():

conn = None

cursor = None

try:

conn = pymysql.connect(

host='localhost',

port=3306,

user='root',

password='你的密码',

database='test_db',

charset='utf8mb4'

)

cursor = conn.cursor()

# 单条插入

sql = "INSERT INTO user (name, age, gender) VALUES (%s, %s, %s)"

# 执行SQL并传入参数

cursor.execute(sql, ("张三", 20, "男"))

# 批量插入（效率更高）

data_list = [

("李四", 22, "男"),

("王五", 19, "女")

]

cursor.executemany(sql, data_list)

# 提交事务

conn.commit()

print(f"插入成功，影响行数：{cursor.rowcount}")

except pymysql.MySQLError as e:

# 出错回滚事务

conn.rollback()

print("插入失败：", e)

finally:

if cursor:

cursor.close()

if conn:

conn.close()

3. 数据更新（UPDATE）

def update_data():

conn = None

cursor = None

try:

conn = pymysql.connect(

host='localhost',

port=3306,

user='root',

password='你的密码',

database='test_db',

charset='utf8mb4'

)

cursor = conn.cursor()

# 更新张三的年龄

sql = "UPDATE user SET age = %s WHERE name = %s"

cursor.execute(sql, (21, "张三"))

conn.commit()

print(f"更新成功，影响行数：{cursor.rowcount}")

except pymysql.MySQLError as e:

conn.rollback()

print("更新失败：", e)

finally:

if cursor:

cursor.close()

if conn:

conn.close()

4. 数据删除（DELETE）

def delete_data():

conn = None

cursor = None

try:

conn = pymysql.connect(

host='localhost',

port=3306,

user='root',

password='你的密码',

database='test_db',

charset='utf8mb4'

)

cursor = conn.cursor()

# 删除年龄小于18的用户

sql = "DELETE FROM user WHERE age < %s"

cursor.execute(sql, (18,))

conn.commit()

print(f"删除成功，影响行数：{cursor.rowcount}")

except pymysql.MySQLError as e:

conn.rollback()

print("删除失败：", e)

finally:

if cursor:

cursor.close()

if conn:

conn.close()

四、关键技巧：避免踩坑

1. 防止SQL注入

绝对不要直接拼接SQL字符串，必须使用参数化查询（如上述示例中的%s占位符）。错误示例（禁止使用）：

# 危险！易被SQL注入

name = "张三' OR 1=1 --"

sql = f"SELECT * FROM user WHERE name = '{name}'"

cursor.execute(sql)

2. 事务处理

MySQL默认开启事务，增删改操作需commit()提交；出错时用rollback()回滚，保证数据一致性。推荐写法：

try:

# 执行多个SQL操作

cursor.execute(sql1)

cursor.execute(sql2)

conn.commit() # 全部成功才提交

except Exception as e:

conn.rollback() # 有错误则回滚

print("事务失败：", e)

3. 字符编码问题

连接时指定charset='utf8mb4'（而非utf8），utf8mb4支持emoji等特殊字符，避免插入数据时出现编码错误。

4. 资源释放

无论操作成功与否，都要关闭游标和连接。推荐使用with语句自动释放资源，简化代码：

import pymysql

# with语句自动关闭连接和游标

with pymysql.connect(

host='localhost',

port=3306,

user='root',

password='你的密码',

database='test_db',

charset='utf8mb4'

) as conn:

with conn.cursor(cursor=pymysql.cursors.DictCursor) as cursor:

sql = "SELECT * FROM user LIMIT 10"

cursor.execute(sql)

results = cursor.fetchall()

print(results)

五、常见问题排查

1. 连接失败：Access denied for user

检查用户名/密码是否正确；

确认MySQL用户是否允许远程连接（如需远程访问，执行GRANT ALL ON *.* TO 'root'@'%' IDENTIFIED BY '密码'; FLUSH PRIVILEGES;）。

2. 连接超时：TimeoutError

检查MySQL服务器是否启动；

确认防火墙/安全组开放3306端口；

增加连接超时参数：connect_timeout=10。

3. 数据插入后查询不到

未执行conn.commit()提交事务；

检查数据库是否选错（database参数）。