Python PyMySQL 数据库操作与 SQL 注入防护实战
环境准备与模块安装
在Python环境中操作MySQL数据库，PyMySQL 是一个纯Python实现的轻量级客户端库。可以通过包管理工具直接安装：
pip install PyMySQL建立数据库连接
与MySQL服务器建立连接是进行任何数据库操作的前提。以下是初始化连接对象的配置示例：
import pymysql
# 初始化数据库连接
db_conn = pymysql.connect(
host="localhost", # 数据库服务器地址
port=3306, # 默认端口
user="db_admin", # 数据库用户名
password="secure_pwd", # 用户密码
db="enterprise_db", # 目标数据库名（也可使用 database 参数）
charset="utf8mb4" # 字符集，推荐使用 utf8mb4 以支持完整 Unicode
)
核心操作：游标与数据检索
在连接建立后，需要创建游标（Cursor）对象来执行SQL语句并获取结果。
1. 创建游标与执行查询
默认情况下，查询结果以元组形式返回。为了提升数据的可读性，通常将其配置为字典格式：
# 创建字典游标，返回结果格式为 [{}, {}]
cur = db_conn.cursor(cursor=pymysql.cursors.DictCursor)
# 定义并执行 SQL 查询
query_sql = "SELECT emp_id, name, department FROM employees WHERE status = 'active'"
affected_count = cur.execute(query_sql)
print(f"查询到 {affected_count} 条有效记录")
2. 提取查询结果
游标提供了多种方法来提取执行后的数据集：
single_row = cur.fetchone() # 获取结果集中的下一条记录
all_rows = cur.fetchall() # 获取结果集中的所有剩余记录
batch_rows = cur.fetchmany(5) # 获取指定数量（如5条）的记录
3. 游标位置控制
在遍历结果集时，游标指针会自动向后移动。若需要重新读取之前的数据，可以使用 scroll 方法调整指针位置：
cur.scroll(2, mode='relative') # 相对于当前位置向后移动2个单位
cur.scroll(0, mode='absolute') # 将指针重置到结果集的起始位置
数据修改与事务管理
对于 SELECT 查询，执行后无需额外操作。但对于 INSERT、UPDATE 和 DELETE 等修改数据的操作，MySQL 默认开启事务，需要手动提交更改。
update_sql = "UPDATE employees SET department = 'AI研发部' WHERE emp_id = 1024"
cur.execute(update_sql)
# 手动提交事务，使修改生效
db_conn.commit()
如果希望简化流程，可以在建立连接时开启自动提交模式：
db_conn = pymysql.connect(
host="localhost",
user="db_admin",
password="secure_pwd",
db="enterprise_db",
charset="utf8mb4",
autocommit=True # 开启自动提交，执行增删改后自动 commit
)
实战：基于数据库的账户验证系统
以下代码展示了一个简单的系统管理员注册与登录验证逻辑：
import pymysql
import hashlib
def get_db_connection():
return pymysql.connect(
host="127.0.0.1", user="root", password="root_pwd",
db="sys_management", charset="utf8mb4", autocommit=True
)
def hash_password(pwd):
return hashlib.sha256(pwd.encode('utf-8')).hexdigest()
def register_admin():
conn = get_db_connection()
cur = conn.cursor()
uname = input("设置管理员账号: ").strip()
# 检查账号是否已存在
cur.execute("SELECT id FROM sys_admins WHERE username = %s", (uname,))
if cur.fetchone():
print("该账号已被注册。")
return
pwd = input("设置登录密码: ").strip()
confirm = input("再次确认密码: ").strip()
if pwd != confirm:
print("两次密码输入不一致。")
return
hashed_pwd = hash_password(pwd)
cur.execute("INSERT INTO sys_admins (username, pass_hash) VALUES (%s, %s)", (uname, hashed_pwd))
print("管理员账号注册成功！")
conn.close()
def verify_admin():
conn = get_db_connection()
cur = conn.cursor()
uname = input("请输入账号: ").strip()
pwd = input("请输入密码: ").strip()
hashed_pwd = hash_password(pwd)
cur.execute("SELECT id FROM sys_admins WHERE username = %s AND pass_hash = %s", (uname, hashed_pwd))
if cur.fetchone():
print("身份验证通过，欢迎进入系统。")
else:
print("账号或密码错误。")
conn.close()
安全隐患：SQL注入攻击
如果在构建SQL语句时直接使用字符串拼接（如 f-string 或 +），恶意用户可以通过构造特殊输入来改变SQL的原始逻辑，这就是SQL注入。
注入场景复现
假设存在以下脆弱的登录验证代码：
uname = input("账号: ")
pwd = input("密码: ")
# 危险的字符串拼接
vulnerable_sql = f"SELECT * FROM sys_admins WHERE username='{uname}' AND pass_hash='{pwd}'"
cur.execute(vulnerable_sql)
攻击者可以尝试以下输入：
- 绕过密码验证：账号输入
admin' --，密码随意。
生成的SQL：SELECT * FROM sys_admins WHERE username='admin' --' AND pass_hash='任意值'
（--注释掉了后续的密码校验） - 无条件登录：账号输入
' OR '1'='1，密码输入' OR '1'='1。
生成的SQL：SELECT * FROM sys_admins WHERE username='' OR '1'='1' AND pass_hash='' OR '1'='1'
（条件恒为真，返回所有记录）
防御策略：参数化查询
PyMySQL 提供了参数化查询机制，通过占位符 %s 将数据与SQL结构分离，由底层驱动自动处理转义，从而彻底杜绝SQL注入。
# 使用 %s 作为占位符，不要加引号
secure_sql = "SELECT * FROM sys_admins WHERE username = %s AND pass_hash = %s"
# 将参数以元组形式传递给 execute 方法
cur.execute(secure_sql, (uname, hashed_pwd))
批量数据操作
当需要插入或更新多条记录时，可以使用 executemany 方法，它不仅安全，而且执行效率更高：
batch_insert_sql = "INSERT INTO system_logs (event_type, description, created_at) VALUES (%s, %s, NOW())"
# 准备多条数据
log_data = [
("LOGIN", "Admin logged in"),
("UPDATE", "Modified user permissions"),
("DELETE", "Removed obsolete records")
]
# 批量执行
cur.executemany(batch_insert_sql, log_data)
print(f"成功写入 {cur.rowcount} 条日志。")
