Python MySQL错误处理完全指南
为什么需要专门的MySQL错误处理？
在Python数据库应用中，完善的错误处理机制至关重要：
- 防止程序因数据库问题而意外终止
- 提供有意义的错误信息帮助调试
- 维护数据完整性和一致性
- 提升应用稳定性和用户体验
- 实现安全的重连机制
常见MySQL错误类型及原因
连接错误
数据库服务未启动、网络问题、认证失败等
语法错误
SQL语句编写错误、无效的表/列名、引号不匹配等
操作错误
违反唯一约束、外键约束、数据类型不匹配等
超时错误
查询执行时间过长、连接闲置超时等
基础错误处理模式
使用try-except块捕获MySQL异常的基本结构：
import pymysql
from pymysql import MySQLError
try:
# 创建数据库连接
connection = pymysql.connect(
host='localhost',
user='user',
password='password',
database='mydb',
charset='utf8mb4',
cursorclass=pymysql.cursors.DictCursor
)
try:
# 创建游标执行操作
with connection.cursor() as cursor:
# 执行SQL语句
sql = "INSERT INTO users (email, password) VALUES (%s, %s)"
cursor.execute(sql, ('user@example.com', 'secure_password'))
# 提交事务
connection.commit()
except MySQLError as e:
# 出错时回滚事务
connection.rollback()
print(f"数据库操作出错: {e}")
finally:
# 确保连接被关闭
connection.close()
except pymysql.err.OperationalError as oe:
print(f"连接数据库失败: {oe}")
except Exception as e:
print(f"发生未预期错误: {e}")
关键点说明：
- 外层try-except捕获连接错误
- 内层try-except捕获执行错误
- 使用特定异常类
MySQLError捕获数据库错误 - 出错时执行回滚操作保证数据一致性
- finally块确保资源释放
高级错误处理技巧
1. 错误重连机制
处理数据库连接断开后的自动重连：
def create_connection(max_retries=3, retry_delay=2):
retries = 0
while retries < max_retries:
try:
return pymysql.connect(
host='localhost',
user='user',
password='password',
database='mydb'
)
except pymysql.err.OperationalError:
retries += 1
print(f"连接失败，尝试重连 ({retries}/{max_retries})")
time.sleep(retry_delay)
raise Exception("无法建立数据库连接")
# 使用示例
try:
connection = create_connection()
# 执行数据库操作...
except Exception as e:
print(f"数据库操作失败: {e}")
2. 错误日志记录
将错误信息记录到日志文件：
import logging
# 配置日志
logging.basicConfig(
filename='db_errors.log',
level=logging.ERROR,
format='%(asctime)s - %(levelname)s - %(message)s'
)
try:
# 数据库操作...
cursor.execute("SELECT * FROM non_existent_table")
except pymysql.err.ProgrammingError as pe:
logging.error(f"SQL语法错误: {pe}")
except pymysql.err.InternalError as ie:
logging.error(f"数据库内部错误: {ie}")
except MySQLError as e:
error_code, error_message = e.args
logging.error(f"MySQL错误 [代码{error_code}]: {error_message}")
3. 处理特定错误代码
根据MySQL错误代码进行特殊处理：
from pymysql.constants.ER import DUP_ENTRY
try:
cursor.execute("INSERT INTO users (username) VALUES ('admin')")
except MySQLError as e:
error_code = e.args[0]
if error_code == DUP_ENTRY: # 1062 - 重复键错误
print("用户名已存在，请选择其他用户名")
elif error_code == 1048: # 列不能为NULL
print("缺少必要字段")
elif error_code == 1452: # 外键约束失败
print("关联数据不存在")
else:
print(f"数据库错误: {e}")
MySQL错误处理最佳实践
- 使用上下文管理器：确保资源正确释放
- 事务粒度控制：合理划分事务范围
- 错误信息友好化：将技术性错误转换为用户可理解信息
- 参数化查询：使用占位符防止SQL注入
- 连接池管理：在高并发应用中使用连接池
- 超时设置：为查询设置合理超时时间
- 定期维护：监控数据库性能并优化查询
完整错误处理示例
结合多种技术的完整数据库操作示例：
import pymysql
from pymysql import MySQLError
import logging
# 配置日志
logging.basicConfig(level=logging.INFO)
def execute_safe_query(query, params=None):
try:
# 创建数据库连接（生产环境应使用连接池）
with pymysql.connect(
host='localhost',
user='user',
password='password',
database='mydb',
connect_timeout=5 # 连接超时设置
) as conn:
# 设置查询执行超时
with conn.cursor() as cursor:
cursor.execute("SET SESSION max_execution_time=3000") # 3秒超时
try:
# 执行查询
cursor.execute(query, params or ())
# 对于SELECT查询获取结果
if query.strip().upper().startswith('SELECT'):
return cursor.fetchall()
# 非查询操作提交事务
conn.commit()
return cursor.rowcount
except MySQLError as e:
conn.rollback()
error_code = e.args[0]
# 处理特定错误
if error_code == 1062:
logging.warning("重复记录错误")
return None
elif error_code == 1205: # 锁等待超时
logging.warning("查询超时，请重试")
return None
else:
logging.error(f"数据库错误: {e}")
raise
except Exception as e:
conn.rollback()
logging.error(f"未知错误: {e}")
raise
except pymysql.err.OperationalError as oe:
logging.error(f"数据库连接失败: {oe}")
return None
except Exception as e:
logging.error(f"系统错误: {e}")
return None
# 使用示例
result = execute_safe_query(
"INSERT INTO orders (user_id, product_id, quantity) VALUES (%s, %s, %s)",
(123, 456, 2)
)
总结
有效的MySQL错误处理是构建健壮Python应用的关键。通过：
- 理解常见的MySQL错误类型
- 使用结构化的try-except块
- 实现错误重连机制
- 记录详细的错误日志
- 针对特定错误代码进行处理
您可以创建出能够优雅处理数据库异常、自动恢复的应用系统，大幅提升用户体验和系统可靠性。
发表评论