《Python中使用MySQLdb连接MySQL》
在当今数据驱动的时代，数据库已成为存储和管理结构化数据的核心工具。MySQL作为开源关系型数据库的代表，凭借其高性能、可靠性和易用性，广泛应用于Web开发、数据分析等领域。而Python作为一门简洁高效的编程语言，与MySQL的结合为开发者提供了强大的数据处理能力。本文将详细介绍如何使用Python的MySQLdb模块（或其替代品PyMySQL）连接MySQL数据库，涵盖连接配置、基础操作、事务管理、错误处理及性能优化等关键内容，帮助读者快速掌握这一技术栈。
一、MySQLdb与PyMySQL的选择
MySQLdb是Python连接MySQL的传统模块，基于C语言开发，性能优异，但仅支持Python 2.x且安装依赖较多。随着Python 3的普及，PyMySQL成为更主流的选择——它纯Python实现，兼容Python 3，安装简单（通过pip即可），且API与MySQLdb高度一致。本文以PyMySQL为例进行讲解，但代码逻辑同样适用于MySQLdb。
二、环境准备与模块安装
在开始前，需确保已安装MySQL服务器并创建测试数据库。以Ubuntu系统为例，安装MySQL和Python模块的步骤如下：
# 安装MySQL服务器
sudo apt update
sudo apt install mysql-server
# 安装PyMySQL
pip install pymysql
安装完成后，可通过以下命令验证MySQL服务状态：
sudo systemctl status mysql
三、建立数据库连接
连接MySQL的核心是创建连接对象，需指定主机、用户名、密码、数据库名等参数。以下是一个基础连接示例：
import pymysql
# 连接参数
config = {
'host': 'localhost', # 数据库主机地址
'user': 'root', # 用户名
'password': 'your_password', # 密码
'database': 'test_db', # 数据库名
'charset': 'utf8mb4', # 字符集
'cursorclass': pymysql.cursors.DictCursor # 返回字典格式结果
}
try:
# 建立连接
connection = pymysql.connect(**config)
print("数据库连接成功！")
except pymysql.Error as e:
print(f"连接失败：{e}")
finally:
# 关闭连接（此处暂不关闭，后续操作使用）
pass
关键参数说明：
-
host：若数据库在远程服务器，需替换为IP或域名。 -
port：默认3306，若修改需显式指定。 -
autocommit：设为True可自动提交事务，默认False需手动提交。 -
cursorclass：使用DictCursor可使查询结果以字典形式返回，键为列名。
四、执行SQL查询
连接建立后，需通过游标（Cursor）对象执行SQL语句。游标提供了execute()、fetchone()、fetchall()等方法。
1. 创建表
create_table_sql = """
CREATE TABLE IF NOT EXISTS users (
id INT AUTO_INCREMENT PRIMARY KEY,
name VARCHAR(50) NOT NULL,
age INT,
email VARCHAR(100) UNIQUE
)
"""
with connection.cursor() as cursor:
cursor.execute(create_table_sql)
connection.commit() # 提交事务
print("表创建成功！")
2. 插入数据
使用参数化查询防止SQL注入：
insert_sql = "INSERT INTO users (name, age, email) VALUES (%s, %s, %s)"
data = ("张三", 25, "zhangsan@example.com")
with connection.cursor() as cursor:
cursor.execute(insert_sql, data)
connection.commit()
print(f"插入成功，影响行数：{cursor.rowcount}")
批量插入示例：
batch_data = [
("李四", 30, "lisi@example.com"),
("王五", 28, "wangwu@example.com")
]
with connection.cursor() as cursor:
cursor.executemany(insert_sql, batch_data)
connection.commit()
3. 查询数据
select_sql = "SELECT * FROM users WHERE age > %s"
with connection.cursor(pymysql.cursors.DictCursor) as cursor:
cursor.execute(select_sql, (20,))
results = cursor.fetchall() # 获取所有结果
for row in results:
print(f"ID: {row['id']}, 姓名: {row['name']}, 年龄: {row['age']}")
分页查询示例：
def get_users_by_page(page, page_size=10):
offset = (page - 1) * page_size
sql = "SELECT * FROM users LIMIT %s OFFSET %s"
with connection.cursor() as cursor:
cursor.execute(sql, (page_size, offset))
return cursor.fetchall()
4. 更新与删除
# 更新
update_sql = "UPDATE users SET age = %s WHERE name = %s"
with connection.cursor() as cursor:
cursor.execute(update_sql, (26, "张三"))
connection.commit()
# 删除
delete_sql = "DELETE FROM users WHERE id = %s"
with connection.cursor() as cursor:
cursor.execute(delete_sql, (1,))
connection.commit()
五、事务管理
事务确保一组操作要么全部成功，要么全部回滚。MySQLdb/PyMySQL默认不自动提交，需显式调用commit()或rollback()。
try:
with connection.cursor() as cursor:
# 操作1：扣款
cursor.execute("UPDATE accounts SET balance = balance - 100 WHERE user_id = 1")
# 操作2：入账（模拟异常）
cursor.execute("UPDATE accounts SET balance = balance + 100 WHERE user_id = 2")
# 假设此处抛出异常
raise ValueError("转账失败")
connection.commit()
except Exception as e:
connection.rollback()
print(f"事务回滚：{e}")
finally:
connection.close()
六、错误处理与连接池
数据库操作可能因网络、权限等问题失败，需捕获pymysql.Error及其子类。
try:
connection = pymysql.connect(**config)
# 执行操作...
except pymysql.MySQLError as e:
if e.args[0] == 1045: # 访问拒绝错误码
print("用户名或密码错误")
else:
print(f"数据库错误：{e}")
except Exception as e:
print(f"未知错误：{e}")
finally:
if 'connection' in locals():
connection.close()
对于高并发场景，连接池可复用连接，减少开销。可使用DBUtils库实现：
from dbutils.pooled_db import PooledDB
pool = PooledDB(
creator=pymysql,
maxconnections=5, # 最大连接数
mincached=2, # 初始化连接数
**config
)
# 从连接池获取连接
conn = pool.connection()
try:
# 执行操作...
pass
finally:
conn.close() # 归还到连接池
七、性能优化建议
- 使用预处理语句：减少SQL解析开销，防止注入。
-
批量操作：
executemany()比多次execute()高效。 - 索引优化：为查询频繁的列创建索引。
- 连接复用：避免频繁创建/关闭连接。
-
限制结果集：使用
LIMIT避免返回过多数据。
八、完整示例：用户管理系统
import pymysql
from dbutils.pooled_db import PooledDB
# 初始化连接池
pool = PooledDB(
creator=pymysql,
maxconnections=5,
host='localhost',
user='root',
password='your_password',
database='test_db',
charset='utf8mb4'
)
class UserManager:
@staticmethod
def add_user(name, age, email):
sql = "INSERT INTO users (name, age, email) VALUES (%s, %s, %s)"
conn = pool.connection()
try:
with conn.cursor() as cursor:
cursor.execute(sql, (name, age, email))
conn.commit()
return True
except Exception as e:
conn.rollback()
print(f"添加用户失败：{e}")
return False
finally:
conn.close()
@staticmethod
def get_users(age_threshold=None):
sql = "SELECT * FROM users"
params = []
if age_threshold is not None:
sql += " WHERE age > %s"
params.append(age_threshold)
conn = pool.connection()
try:
with conn.cursor(pymysql.cursors.DictCursor) as cursor:
cursor.execute(sql, params)
return cursor.fetchall()
finally:
conn.close()
# 测试
if __name__ == "__main__":
UserManager.add_user("赵六", 35, "zhaoliu@example.com")
users = UserManager.get_users(30)
for user in users:
print(user)
九、总结与扩展
本文详细介绍了Python通过PyMySQL连接MySQL的全流程，包括环境配置、CRUD操作、事务管理、错误处理及性能优化。实际开发中，还可结合ORM框架（如SQLAlchemy）进一步提升效率。对于大规模应用，需考虑分库分表、读写分离等高级架构。
关键词：Python、MySQLdb、PyMySQL、数据库连接、SQL操作、事务管理、连接池、性能优化
简介：本文系统讲解了Python中使用MySQLdb/PyMySQL模块连接MySQL数据库的方法，涵盖连接配置、CRUD操作、事务处理、错误捕获及性能优化技巧，并提供完整代码示例，适合Python开发者快速掌握数据库交互技术。
