1. Python连接MySQL的基础环境配置在开始Python操作MySQL之前我们需要先准备好开发环境。这里我推荐使用Python 3.6版本和MySQL 5.7/8.0的组合这是目前最稳定且广泛使用的搭配。1.1 安装MySQL驱动Python连接MySQL最常用的驱动是PyMySQL和mysql-connector-python。PyMySQL是纯Python实现的MySQL客户端而mysql-connector-python是MySQL官方提供的连接器。我个人更推荐PyMySQL因为它安装简单且兼容性更好。安装PyMySQL非常简单pip install pymysql如果你更倾向于使用官方驱动pip install mysql-connector-python注意在Windows系统上安装mysql-connector-python时可能会遇到依赖问题。如果安装失败可以尝试先安装Microsoft Visual C Build Tools。1.2 数据库连接配置无论使用哪种驱动连接MySQL数据库的基本参数都相同主机地址(host)端口号(port默认3306)用户名(user)密码(password)数据库名称(database)这里是一个标准的连接示例import pymysql # 创建数据库连接 connection pymysql.connect( hostlocalhost, # 数据库服务器地址 userroot, # 用户名 password123456, # 密码 databasetest, # 数据库名 port3306, # 端口默认3306 charsetutf8mb4, # 字符集 cursorclasspymysql.cursors.DictCursor # 返回字典格式的结果 )在实际项目中我建议将这些配置信息存储在配置文件中或环境变量中而不是硬编码在代码里。2. 数据库基本操作CRUD实现CRUD代表Create(创建)、Read(读取)、Update(更新)和Delete(删除)是数据库最基本的四种操作。下面我们分别来看Python中如何实现这些操作。2.1 创建数据(Create)向数据库插入新记录是最基础的操作。我们先创建一个示例表CREATE TABLE users ( id int(11) NOT NULL AUTO_INCREMENT, name varchar(50) NOT NULL, email varchar(100) NOT NULL, created_at timestamp NOT NULL DEFAULT CURRENT_TIMESTAMP, PRIMARY KEY (id) ) ENGINEInnoDB DEFAULT CHARSETutf8mb4;Python插入数据的代码def insert_user(name, email): try: with connection.cursor() as cursor: # 编写SQL语句 sql INSERT INTO users (name, email) VALUES (%s, %s) # 执行SQL cursor.execute(sql, (name, email)) # 提交事务 connection.commit() print(插入成功) except Exception as e: # 发生错误时回滚 connection.rollback() print(f插入失败: {e})重要提示一定要使用参数化查询(%s占位符)而不是字符串拼接这是防止SQL注入攻击的关键措施。2.2 查询数据(Read)查询是最常用的数据库操作。PyMySQL支持多种查询方式2.2.1 基本查询def get_all_users(): try: with connection.cursor() as cursor: sql SELECT id, name, email FROM users cursor.execute(sql) # 获取所有记录 result cursor.fetchall() return result except Exception as e: print(f查询失败: {e}) return None2.2.2 条件查询def get_user_by_id(user_id): try: with connection.cursor() as cursor: sql SELECT id, name, email FROM users WHERE id %s cursor.execute(sql, (user_id,)) # 获取单条记录 result cursor.fetchone() return result except Exception as e: print(f查询失败: {e}) return None2.2.3 分页查询def get_users_paginated(page1, per_page10): try: with connection.cursor() as cursor: offset (page - 1) * per_page sql SELECT id, name, email FROM users LIMIT %s OFFSET %s cursor.execute(sql, (per_page, offset)) result cursor.fetchall() return result except Exception as e: print(f分页查询失败: {e}) return None2.3 更新数据(Update)更新操作需要特别注意条件避免误更新大量数据def update_user_email(user_id, new_email): try: with connection.cursor() as cursor: sql UPDATE users SET email %s WHERE id %s affected_rows cursor.execute(sql, (new_email, user_id)) connection.commit() print(f更新了{affected_rows}条记录) return affected_rows except Exception as e: connection.rollback() print(f更新失败: {e}) return 02.4 删除数据(Delete)删除操作是最危险的操作务必谨慎def delete_user(user_id): try: with connection.cursor() as cursor: sql DELETE FROM users WHERE id %s affected_rows cursor.execute(sql, (user_id,)) connection.commit() print(f删除了{affected_rows}条记录) return affected_rows except Exception as e: connection.rollback() print(f删除失败: {e}) return 03. 高级操作与性能优化掌握了基本的CRUD操作后我们来看一些更高级的用法和性能优化技巧。3.1 批量操作当需要插入大量数据时单条插入效率很低。PyMySQL支持批量操作def batch_insert_users(user_list): try: with connection.cursor() as cursor: sql INSERT INTO users (name, email) VALUES (%s, %s) # 使用executemany进行批量插入 affected_rows cursor.executemany(sql, user_list) connection.commit() print(f批量插入了{affected_rows}条记录) return affected_rows except Exception as e: connection.rollback() print(f批量插入失败: {e}) return 0实测表明批量插入比单条插入快10-100倍特别是在网络延迟较高的情况下。3.2 事务处理事务是保证数据一致性的重要机制。PyMySQL默认开启了自动提交模式但我们可以手动控制事务def transfer_money(from_id, to_id, amount): try: with connection.cursor() as cursor: # 开始事务 connection.begin() # 扣款 sql1 UPDATE accounts SET balance balance - %s WHERE id %s AND balance %s affected1 cursor.execute(sql1, (amount, from_id, amount)) if affected1 0: raise Exception(扣款失败余额不足) # 收款 sql2 UPDATE accounts SET balance balance %s WHERE id %s affected2 cursor.execute(sql2, (amount, to_id)) # 提交事务 connection.commit() print(转账成功) return True except Exception as e: connection.rollback() print(f转账失败: {e}) return False3.3 连接池管理在高并发应用中频繁创建和关闭数据库连接会严重影响性能。可以使用DBUtils等连接池工具from dbutils.pooled_db import PooledDB # 创建连接池 pool PooledDB( creatorpymysql, maxconnections10, # 连接池最大连接数 mincached2, # 初始化时创建的连接数 hostlocalhost, userroot, password123456, databasetest, charsetutf8mb4 ) # 从连接池获取连接 def get_users_with_pool(): conn pool.connection() try: with conn.cursor() as cursor: sql SELECT * FROM users cursor.execute(sql) return cursor.fetchall() finally: conn.close() # 将连接返回连接池4. 常见问题与调试技巧在实际开发中我们经常会遇到各种问题。下面分享一些常见问题的解决方法。4.1 连接问题排查问题1无法连接到MySQL服务器可能原因MySQL服务未启动防火墙阻止了连接用户权限不足连接参数错误解决方法import socket def check_mysql_connection(host, port3306): try: # 检查端口是否开放 sock socket.socket(socket.AF_INET, socket.SOCK_STREAM) sock.settimeout(3) result sock.connect_ex((host, port)) sock.close() return result 0 except Exception as e: print(f连接检查失败: {e}) return False问题2字符集问题导致乱码解决方案确保连接时指定了正确的字符集推荐utf8mb4并且数据库、表和字段都使用相同的字符集。4.2 性能优化建议索引优化为常用查询条件添加索引ALTER TABLE users ADD INDEX idx_email (email);查询优化只查询需要的字段避免SELECT *# 不好的写法 cursor.execute(SELECT * FROM users) # 好的写法 cursor.execute(SELECT id, name FROM users)批量操作使用executemany代替循环执行单条SQL连接复用使用连接池管理数据库连接4.3 错误处理最佳实践良好的错误处理能让你的应用更健壮。PyMySQL常见的异常包括pymysql.err.OperationalError操作错误如连接失败pymysql.err.ProgrammingErrorSQL语法错误pymysql.err.IntegrityError完整性约束错误如重复主键建议的错误处理模式def safe_db_operation(): try: with connection.cursor() as cursor: # 数据库操作代码 pass connection.commit() except pymysql.err.OperationalError as e: print(f数据库操作错误: {e}) connection.rollback() except pymysql.err.ProgrammingError as e: print(fSQL语法错误: {e}) connection.rollback() except pymysql.err.IntegrityError as e: print(f数据完整性错误: {e}) connection.rollback() except Exception as e: print(f未知错误: {e}) connection.rollback()4.4 调试技巧查看实际执行的SQL# 在cursor.execute()之后 print(cursor._last_executed)性能分析import time start time.time() # 执行数据库操作 end time.time() print(f操作耗时: {end - start:.3f}秒)使用EXPLAIN分析查询def explain_query(sql): with connection.cursor() as cursor: cursor.execute(fEXPLAIN {sql}) return cursor.fetchall()5. 实际项目中的应用示例让我们通过一个完整的示例来展示如何在真实项目中使用Python操作MySQL。5.1 用户管理系统实现class UserManager: def __init__(self, db_config): self.connection pymysql.connect(**db_config) def __del__(self): if hasattr(self, connection) and self.connection: self.connection.close() def create_user(self, name, email, password): try: with self.connection.cursor() as cursor: # 检查邮箱是否已存在 if self.get_user_by_email(email): return False, 邮箱已被注册 # 密码应该加密存储 hashed_password self._hash_password(password) sql INSERT INTO users (name, email, password_hash) VALUES (%s, %s, %s) cursor.execute(sql, (name, email, hashed_password)) self.connection.commit() return True, 注册成功 except Exception as e: self.connection.rollback() return False, f注册失败: {str(e)} def get_user_by_email(self, email): try: with self.connection.cursor() as cursor: sql SELECT * FROM users WHERE email %s cursor.execute(sql, (email,)) return cursor.fetchone() except Exception as e: print(f查询用户失败: {e}) return None def authenticate(self, email, password): user self.get_user_by_email(email) if not user: return False, 用户不存在 if not self._verify_password(password, user[password_hash]): return False, 密码错误 return True, 登录成功 def _hash_password(self, password): # 实际项目中应该使用bcrypt等安全哈希算法 import hashlib return hashlib.sha256(password.encode()).hexdigest() def _verify_password(self, password, hashed): return self._hash_password(password) hashed5.2 数据库迁移脚本在实际项目中我们经常需要管理数据库变更。下面是一个简单的迁移脚本示例class MigrationTool: def __init__(self, db_config): self.connection pymysql.connect(**db_config) def apply_migrations(self): self._create_migrations_table() applied self._get_applied_migrations() migrations self._find_pending_migrations(applied) for mig in migrations: try: self._apply_migration(mig) self._record_migration(mig) print(f成功应用迁移: {mig}) except Exception as e: print(f应用迁移{mig}失败: {e}) break def _create_migrations_table(self): sql CREATE TABLE IF NOT EXISTS migrations ( id INT AUTO_INCREMENT PRIMARY KEY, name VARCHAR(255) NOT NULL, applied_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP ) ENGINEInnoDB DEFAULT CHARSETutf8mb4 with self.connection.cursor() as cursor: cursor.execute(sql) self.connection.commit() def _get_applied_migrations(self): with self.connection.cursor() as cursor: cursor.execute(SELECT name FROM migrations ORDER BY id) return {row[name] for row in cursor.fetchall()} def _find_pending_migrations(self, applied): # 实际项目中可以从文件系统读取迁移脚本 all_migrations { 001_create_users_table, 002_add_user_status_column, 003_create_posts_table } return sorted(all_migrations - applied) def _apply_migration(self, migration_name): # 实际项目中应该从文件加载SQL if migration_name 001_create_users_table: sql CREATE TABLE users ( id INT AUTO_INCREMENT PRIMARY KEY, name VARCHAR(50) NOT NULL, email VARCHAR(100) NOT NULL UNIQUE, created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP ) ENGINEInnoDB DEFAULT CHARSETutf8mb4 elif migration_name 002_add_user_status_column: sql ALTER TABLE users ADD COLUMN status ENUM(active,inactive) DEFAULT active elif migration_name 003_create_posts_table: sql CREATE TABLE posts ( id INT AUTO_INCREMENT PRIMARY KEY, user_id INT NOT NULL, title VARCHAR(255) NOT NULL, content TEXT, created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP, FOREIGN KEY (user_id) REFERENCES users(id) ) ENGINEInnoDB DEFAULT CHARSETutf8mb4 else: raise ValueError(f未知迁移: {migration_name}) with self.connection.cursor() as cursor: cursor.execute(sql) self.connection.commit() def _record_migration(self, migration_name): with self.connection.cursor() as cursor: sql INSERT INTO migrations (name) VALUES (%s) cursor.execute(sql, (migration_name,)) self.connection.commit()5.3 使用ORM框架虽然直接使用PyMySQL可以满足基本需求但在大型项目中使用ORM(Object-Relational Mapping)框架如SQLAlchemy或Django ORM会更高效# SQLAlchemy示例 from sqlalchemy import create_engine, Column, Integer, String from sqlalchemy.ext.declarative import declarative_base from sqlalchemy.orm import sessionmaker Base declarative_base() class User(Base): __tablename__ users id Column(Integer, primary_keyTrue) name Column(String(50)) email Column(String(100), uniqueTrue) # 创建引擎和会话 engine create_engine(mysqlpymysql://root:123456localhost/test) Session sessionmaker(bindengine) session Session() # 使用ORM操作 new_user User(name张三, emailzhangsanexample.com) session.add(new_user) session.commit() # 查询 users session.query(User).filter(User.name.like(张%)).all() for user in users: print(user.name, user.email)ORM框架虽然会增加一些学习成本但它能显著提高开发效率特别是在复杂的数据关系操作中。