1. Python与MySQL交互基础Python作为当今最流行的编程语言之一与MySQL数据库的交互能力是其核心应用场景。在实际开发中我们通常使用专门的数据库驱动来实现这一功能。目前主流的Python MySQL驱动有以下几种选择MySQL Connector/PythonMySQL官方提供的纯Python驱动PyMySQL纯Python实现的MySQL客户端MySQLdbC语言实现的传统驱动仅支持Python 2aiomysql支持异步IO的MySQL驱动对于大多数应用场景我推荐使用MySQL Connector/Python。这是MySQL官方维护的驱动具有以下优势纯Python实现无需编译安装完全兼容MySQL协议支持Python 3.x提供完善的文档和社区支持安装MySQL Connector非常简单使用pip即可完成pip install mysql-connector-python注意如果同时安装了多个Python版本请确保使用对应版本的pip进行安装。可以通过python -m pip install mysql-connector-python指定Python解释器。2. 数据库连接与基本操作2.1 建立数据库连接建立与MySQL数据库的连接是任何数据库操作的第一步。以下是一个完整的连接示例import mysql.connector config { host: localhost, user: your_username, password: your_password, database: your_database, port: 3306, charset: utf8mb4, connection_timeout: 10 } try: conn mysql.connector.connect(**config) print(数据库连接成功) except mysql.connector.Error as err: print(f连接失败: {err}) finally: if conn in locals() and conn.is_connected(): conn.close()在实际项目中我建议将数据库配置信息存储在环境变量或配置文件中而不是直接硬编码在代码里。这样可以提高安全性也便于不同环境的切换。2.2 执行SQL查询执行SQL查询是数据库操作的核心。MySQL Connector提供了两种主要方式使用cursor.execute()执行简单查询使用cursor.executemany()执行批量操作# 查询示例 cursor conn.cursor(dictionaryTrue) # 返回字典形式的结果 cursor.execute(SELECT * FROM users WHERE age %s, (18,)) results cursor.fetchall() for row in results: print(row[username], row[email]) # 插入示例 insert_sql INSERT INTO users (username, email, age) VALUES (%s, %s, %s) user_data (john_doe, johnexample.com, 25) cursor.execute(insert_sql, user_data) conn.commit() # 重要必须提交事务重要提示始终使用参数化查询如上例中的%s占位符而不是字符串拼接这是防止SQL注入攻击的关键措施。3. 高级操作与性能优化3.1 事务处理MySQL支持事务操作这对于保证数据一致性至关重要。Python中处理事务的典型模式如下try: conn.start_transaction() cursor.execute(UPDATE accounts SET balance balance - %s WHERE id %s, (100, 1)) cursor.execute(UPDATE accounts SET balance balance %s WHERE id %s, (100, 2)) conn.commit() except Exception as e: conn.rollback() print(f事务失败: {e})3.2 批量操作当需要处理大量数据时批量操作可以显著提高性能users [ (user1, user1example.com, 20), (user2, user2example.com, 22), (user3, user3example.com, 25) ] cursor.executemany( INSERT INTO users (username, email, age) VALUES (%s, %s, %s), users ) conn.commit()3.3 连接池管理对于Web应用等高频访问数据库的场景使用连接池是必要的from mysql.connector import pooling dbconfig { host: localhost, user: your_username, password: your_password, database: your_db } connection_pool pooling.MySQLConnectionPool( pool_namemypool, pool_size5, **dbconfig ) # 使用连接池 conn connection_pool.get_connection() cursor conn.cursor() # 执行操作... conn.close() # 实际是返回到连接池4. 实战项目用户管理系统让我们通过一个完整的用户管理系统示例综合运用上述知识import mysql.connector from mysql.connector import Error class UserManager: def __init__(self): self.connection None try: self.connection mysql.connector.connect( hostlocalhost, userroot, passwordpassword, databaseuser_management ) except Error as e: print(f连接数据库失败: {e}) raise def create_user(self, username, email, age): try: cursor self.connection.cursor() cursor.execute( INSERT INTO users (username, email, age) VALUES (%s, %s, %s), (username, email, age) ) self.connection.commit() return cursor.lastrowid except Error as e: self.connection.rollback() print(f创建用户失败: {e}) return None def get_users(self, min_ageNone): try: cursor self.connection.cursor(dictionaryTrue) if min_age: cursor.execute( SELECT * FROM users WHERE age %s ORDER BY username, (min_age,) ) else: cursor.execute(SELECT * FROM users ORDER BY username) return cursor.fetchall() except Error as e: print(f获取用户列表失败: {e}) return [] def update_user(self, user_id, **kwargs): if not kwargs: return False set_clause , .join([f{k} %s for k in kwargs]) values list(kwargs.values()) values.append(user_id) try: cursor self.connection.cursor() cursor.execute( fUPDATE users SET {set_clause} WHERE id %s, values ) self.connection.commit() return cursor.rowcount 0 except Error as e: self.connection.rollback() print(f更新用户失败: {e}) return False def __del__(self): if self.connection and self.connection.is_connected(): self.connection.close() # 使用示例 if __name__ __main__: manager UserManager() # 创建用户 new_user_id manager.create_user(test_user, testexample.com, 30) print(f新用户ID: {new_user_id}) # 查询用户 users manager.get_users(min_age18) for user in users: print(user) # 更新用户 success manager.update_user(new_user_id, emailnew_emailexample.com, age31) print(f更新结果: {成功 if success else 失败})5. 常见问题与解决方案5.1 连接超时问题MySQL服务器默认会在8小时不活动后关闭连接。解决方法在连接配置中添加pool_reset_sessionFalse使用连接池并设置合理的连接超时时间捕获连接错误并自动重连config { # ...其他配置 pool_reset_session: False, pool_size: 5, pool_name: mypool, pool_timeout: 30 }5.2 字符编码问题确保数据库、表和连接都使用UTF-8编码# 创建连接时指定字符集 conn mysql.connector.connect( # ...其他参数 charsetutf8mb4, collationutf8mb4_unicode_ci ) # 创建表时指定字符集 CREATE TABLE users ( id INT AUTO_INCREMENT PRIMARY KEY, username VARCHAR(50), email VARCHAR(100) ) CHARACTER SET utf8mb4 COLLATE utf8mb4_unicode_ci;5.3 性能优化技巧使用索引加速查询合理使用批量操作代替循环单条插入只查询需要的列避免SELECT *对于复杂查询考虑使用存储过程定期分析慢查询日志# 添加索引示例 cursor.execute( CREATE INDEX idx_users_age ON users(age) ) # 使用EXPLAIN分析查询 cursor.execute(EXPLAIN SELECT * FROM users WHERE age 18) print(cursor.fetchall())在实际开发中我发现连接管理和错误处理是最容易出问题的环节。建议封装一个基础的数据库操作类统一处理连接、事务和异常避免重复代码。同时对于生产环境的应用一定要实现完善的日志记录便于排查问题。