MySQL初级篇（四）：Python实战CRUD与事务安全 1. Python连接MySQL的准备工作在开始操作MySQL之前我们需要先准备好Python环境。这里我推荐使用pymysql这个库它是Python中最常用的MySQL驱动之一。安装起来非常简单只需要一条命令pip install pymysql安装完成后我们就可以在Python代码中导入这个模块了。不过在正式连接数据库之前我建议你先确认MySQL服务已经启动并且知道以下信息数据库服务器的IP地址本地通常是127.0.0.1数据库端口号默认是3306数据库用户名和密码要连接的数据库名称我曾经遇到过不少初学者在这一步卡住最常见的问题就是忘记启动MySQL服务或者输错了密码。所以建议你先用命令行工具测试下能否连接成功再写Python代码。2. 数据库连接与基本操作2.1 建立数据库连接连接MySQL数据库的代码其实很简单但有几个细节需要注意。下面是一个完整的连接示例import pymysql def get_db_connection(): try: conn pymysql.connect( host127.0.0.1, userroot, passwordyourpassword, databasetest_db, port3306, charsetutf8mb4, cursorclasspymysql.cursors.DictCursor ) print(数据库连接成功) return conn except pymysql.MySQLError as e: print(f数据库连接失败{e}) return None这里有几个关键点需要注意charset最好设置为utf8mb4这样才能支持完整的Unicode字符包括emojicursorclass设置为DictCursor可以让查询结果以字典形式返回更方便处理一定要用try-except捕获异常数据库连接失败是很常见的情况2.2 执行SQL查询有了连接对象后我们就可以执行SQL语句了。先来看一个简单的查询例子def query_data(conn): try: with conn.cursor() as cursor: sql SELECT * FROM users WHERE age %s cursor.execute(sql, (18,)) results cursor.fetchall() for row in results: print(row) except pymysql.MySQLError as e: print(f查询失败{e})这里有几个最佳实践使用with语句管理cursor这样不需要手动关闭SQL语句中使用%s作为占位符不要直接拼接字符串防止SQL注入参数以元组形式传给execute方法2.3 插入、更新和删除数据对于写操作INSERT/UPDATE/DELETE流程和查询类似但需要注意事务处理def insert_user(conn, name, age): try: with conn.cursor() as cursor: sql INSERT INTO users (name, age) VALUES (%s, %s) cursor.execute(sql, (name, age)) conn.commit() # 重要必须提交事务 print(插入数据成功) except pymysql.MySQLError as e: conn.rollback() # 出错时回滚 print(f插入数据失败{e})更新和删除操作的代码结构类似只是SQL语句不同。记住写操作后一定要调用commit()否则更改不会保存到数据库。3. 事务处理与ACID特性3.1 什么是事务事务是数据库操作的基本单位它保证一组操作要么全部成功要么全部失败。MySQL的事务有四个特性ACID原子性(Atomicity)事务是不可分割的工作单位一致性(Consistency)事务执行前后数据库从一个一致状态变到另一个一致状态隔离性(Isolation)多个事务并发执行时一个事务的执行不应影响其他事务持久性(Durability)一旦事务提交其结果就是永久性的3.2 Python中的事务处理在Python中我们可以这样处理事务def transfer_money(conn, from_id, to_id, amount): try: with conn.cursor() as cursor: # 检查转出账户余额 cursor.execute(SELECT balance FROM accounts WHERE id%s, (from_id,)) from_balance cursor.fetchone()[balance] if from_balance amount: raise ValueError(余额不足) # 执行转账 cursor.execute(UPDATE accounts SET balancebalance-%s WHERE id%s, (amount, from_id)) cursor.execute(UPDATE accounts SET balancebalance%s WHERE id%s, (amount, to_id)) conn.commit() print(转账成功) except Exception as e: conn.rollback() print(f转账失败{e})这个例子展示了典型的转账场景要么两个账户都更新成功要么都不更新。如果在两个UPDATE之间发生错误事务会回滚保证数据一致性。3.3 设置事务隔离级别MySQL支持多种事务隔离级别可以通过Python设置def set_isolation_level(conn, level): levels { READ UNCOMMITTED: SET SESSION TRANSACTION ISOLATION LEVEL READ UNCOMMITTED, READ COMMITTED: SET SESSION TRANSACTION ISOLATION LEVEL READ COMMITTED, REPEATABLE READ: SET SESSION TRANSACTION ISOLATION LEVEL REPEATABLE READ, SERIALIZABLE: SET SESSION TRANSACTION ISOLATION LEVEL SERIALIZABLE } try: with conn.cursor() as cursor: cursor.execute(levels[level]) print(f已设置隔离级别为{level}) except pymysql.MySQLError as e: print(f设置隔离级别失败{e})不同的隔离级别对性能和数据一致性有不同的影响需要根据业务需求选择。4. 防范SQL注入攻击4.1 什么是SQL注入SQL注入是一种常见的安全漏洞攻击者通过在输入中插入恶意SQL代码欺骗服务器执行非预期的SQL命令。比如# 危险不要这样做 user_input admin; DROP TABLE users; -- sql fSELECT * FROM users WHERE username{user_input} cursor.execute(sql)这段代码可能导致users表被删除4.2 使用参数化查询防范SQL注入最简单有效的方法就是使用参数化查询# 安全的方式 user_input admin; DROP TABLE users; -- sql SELECT * FROM users WHERE username%s cursor.execute(sql, (user_input,))pymysql会自动处理参数确保输入值被正确转义。4.3 其他安全措施除了参数化查询外还有一些额外的安全建议最小权限原则数据库用户只授予必要的最小权限输入验证在应用层验证用户输入使用ORM框架如SQLAlchemy等它们通常内置了防注入机制定期更新保持pymysql和MySQL的版本最新5. 性能优化技巧5.1 使用连接池频繁创建和关闭数据库连接很耗资源可以使用连接池from pymysql import pools # 创建连接池 pool pools.Pool(host127.0.0.1, userroot, passwordyourpassword, databasetest_db, port3306, charsetutf8mb4, min_size3, max_size10) def get_from_pool(): return pool.connection()连接池会自动管理连接提高性能。5.2 批量操作对于大量数据操作使用批量插入比单条插入高效得多def batch_insert(conn, data): try: with conn.cursor() as cursor: sql INSERT INTO users (name, age) VALUES (%s, %s) cursor.executemany(sql, data) conn.commit() print(f批量插入了{len(data)}条数据) except pymysql.MySQLError as e: conn.rollback() print(f批量插入失败{e})5.3 使用索引确保查询字段有适当的索引可以大幅提高查询速度。虽然索引是在数据库层面设置的但Python代码中可以检测查询性能def explain_query(conn, sql): try: with conn.cursor() as cursor: cursor.execute(fEXPLAIN {sql}) result cursor.fetchall() for row in result: print(row) except pymysql.MySQLError as e: print(f执行EXPLAIN失败{e})EXPLAIN命令可以帮助分析查询执行计划找出性能瓶颈。6. 实际项目中的经验分享在实际项目中我总结了一些有用的经验连接管理确保每个请求结束后都正确关闭连接可以使用上下文管理器from contextlib import contextmanager contextmanager def db_connection(): conn None try: conn get_db_connection() yield conn finally: if conn: conn.close()错误处理区分不同类型的数据库错误采取不同策略try: # 数据库操作 except pymysql.err.IntegrityError as e: # 处理唯一约束违反等完整性错误 except pymysql.err.OperationalError as e: # 处理连接问题等操作错误 except pymysql.err.ProgrammingError as e: # 处理SQL语法错误等编程错误日志记录记录重要的数据库操作方便排查问题import logging logging.basicConfig(levellogging.INFO) logger logging.getLogger(__name__) def log_query(sql, paramsNone): logger.info(f执行SQL: {sql}) if params: logger.info(f参数: {params})超时设置为长时间运行的查询设置超时conn pymysql.connect( # 其他参数... read_timeout10, # 秒 write_timeout10 )连接重试网络不稳定时实现自动重连import time def safe_query(conn, sql, paramsNone, max_retries3): for attempt in range(max_retries): try: with conn.cursor() as cursor: cursor.execute(sql, params) return cursor.fetchall() except pymysql.err.OperationalError as e: if attempt max_retries - 1: raise time.sleep(1) conn.ping(reconnectTrue) # 重新连接