Python连接MySQL数据库的完整指南与最佳实践

1. Python连接MySQL的基础准备在开始操作数据库之前我们需要先完成基础环境的搭建。Python连接MySQL最常用的库是PyMySQL和mysql-connector-python我个人更推荐PyMySQL因为它纯Python实现兼容性更好安装也更简单。首先安装PyMySQLpip install pymysql如果是生产环境建议指定版本号pip install pymysql1.0.2注意如果同时安装了多个Python版本要确认pip命令对应的是你使用的Python版本。可以使用python -m pip install pymysql来避免版本混淆。2. 建立数据库连接连接MySQL数据库的基本代码如下import pymysql # 创建连接 connection pymysql.connect( hostlocalhost, # 数据库服务器地址 userusername, # 用户名 passwordpassword, # 密码 databasetest_db, # 数据库名 port3306, # 端口默认3306 charsetutf8mb4, # 字符集 cursorclasspymysql.cursors.DictCursor # 返回字典格式的结果 ) # 使用连接 try: with connection.cursor() as cursor: # 执行SQL语句 sql SELECT * FROM users cursor.execute(sql) result cursor.fetchall() print(result) finally: connection.close() # 关闭连接在实际项目中我建议将数据库连接信息放在配置文件中而不是硬编码在代码里。可以创建一个config.py文件# config.py DB_CONFIG { host: localhost, user: username, password: password, database: test_db, charset: utf8mb4, cursorclass: pymysql.cursors.DictCursor }然后在主程序中导入使用from config import DB_CONFIG connection pymysql.connect(**DB_CONFIG)3. 数据库增删改查操作详解3.1 创建表在操作数据前通常需要先创建表。下面是一个创建用户表的示例def create_user_table(): connection pymysql.connect(**DB_CONFIG) try: with connection.cursor() as cursor: sql CREATE TABLE IF NOT EXISTS users ( id INT AUTO_INCREMENT PRIMARY KEY, username VARCHAR(50) NOT NULL UNIQUE, email VARCHAR(100) NOT NULL UNIQUE, password VARCHAR(100) NOT NULL, created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP, updated_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP ON UPDATE CURRENT_TIMESTAMP ) ENGINEInnoDB DEFAULT CHARSETutf8mb4 COLLATEutf8mb4_unicode_ci cursor.execute(sql) connection.commit() finally: connection.close()专业提示使用IF NOT EXISTS可以避免表已存在时报错。ENGINEInnoDB指定存储引擎支持事务。utf8mb4字符集可以完整支持emoji等特殊字符。3.2 插入数据增插入数据有几种方式最简单的是直接执行INSERT语句def add_user(username, email, password): connection pymysql.connect(**DB_CONFIG) try: with connection.cursor() as cursor: sql INSERT INTO users (username, email, password) VALUES (%s, %s, %s) cursor.execute(sql, (username, email, password)) connection.commit() return cursor.lastrowid # 返回插入的ID except pymysql.err.IntegrityError as e: # 处理唯一约束冲突 if Duplicate entry in str(e): raise ValueError(用户名或邮箱已存在) raise finally: connection.close()批量插入数据可以使用executemany()方法效率更高def batch_add_users(users): users是包含多个用户信息的元组列表 connection pymysql.connect(**DB_CONFIG) try: with connection.cursor() as cursor: sql INSERT INTO users (username, email, password) VALUES (%s, %s, %s) cursor.executemany(sql, users) connection.commit() return cursor.rowcount # 返回插入的行数 finally: connection.close()3.3 查询数据查基础查询非常简单def get_all_users(): connection pymysql.connect(**DB_CONFIG) try: with connection.cursor() as cursor: sql SELECT * FROM users cursor.execute(sql) return cursor.fetchall() # 获取所有记录 finally: connection.close()带条件的查询def get_user_by_id(user_id): connection pymysql.connect(**DB_CONFIG) try: with connection.cursor() as cursor: sql SELECT * FROM users WHERE id %s cursor.execute(sql, (user_id,)) return cursor.fetchone() # 获取单条记录 finally: connection.close()分页查询def get_users_paginated(page1, per_page10): connection pymysql.connect(**DB_CONFIG) try: with connection.cursor() as cursor: sql SELECT * FROM users LIMIT %s OFFSET %s offset (page - 1) * per_page cursor.execute(sql, (per_page, offset)) return cursor.fetchall() finally: connection.close()3.4 更新数据改更新数据的基本模式def update_user_password(user_id, new_password): connection pymysql.connect(**DB_CONFIG) try: with connection.cursor() as cursor: sql UPDATE users SET password %s WHERE id %s cursor.execute(sql, (new_password, user_id)) connection.commit() return cursor.rowcount # 返回受影响的行数 finally: connection.close()批量更新def batch_update_users(updates): updates是包含(id, new_password)的元组列表 connection pymysql.connect(**DB_CONFIG) try: with connection.cursor() as cursor: sql UPDATE users SET password %s WHERE id %s cursor.executemany(sql, updates) connection.commit() return cursor.rowcount finally: connection.close()3.5 删除数据删删除单条记录def delete_user(user_id): connection pymysql.connect(**DB_CONFIG) try: with connection.cursor() as cursor: sql DELETE FROM users WHERE id %s cursor.execute(sql, (user_id,)) connection.commit() return cursor.rowcount finally: connection.close()批量删除def delete_users_by_ids(user_ids): connection pymysql.connect(**DB_CONFIG) try: with connection.cursor() as cursor: # 使用IN语句和元组展开 sql DELETE FROM users WHERE id IN ({}).format( ,.join([%s] * len(user_ids)) ) cursor.execute(sql, tuple(user_ids)) connection.commit() return cursor.rowcount finally: connection.close()4. 高级操作与最佳实践4.1 使用事务对于需要原子性执行的一组操作应该使用事务def transfer_funds(from_id, to_id, amount): connection pymysql.connect(**DB_CONFIG) try: with connection.cursor() as cursor: # 开始事务 connection.begin() # 检查转出账户余额 cursor.execute(SELECT balance FROM accounts WHERE id %s FOR UPDATE, (from_id,)) from_balance cursor.fetchone()[balance] if from_balance amount: raise ValueError(余额不足) # 执行转账 cursor.execute(UPDATE accounts SET balance balance - %s WHERE id %s, (amount, from_id)) cursor.execute(UPDATE accounts SET balance balance %s WHERE id %s, (amount, to_id)) # 提交事务 connection.commit() return True except Exception as e: # 发生错误回滚 connection.rollback() raise finally: connection.close()重要提示使用FOR UPDATE锁定行防止并发修改。事务中的操作要么全部成功要么全部失败。4.2 使用连接池在高并发场景下频繁创建和关闭连接会影响性能。可以使用连接池from dbutils.pooled_db import PooledDB # 创建连接池 pool PooledDB( creatorpymysql, maxconnections10, # 最大连接数 mincached2, # 初始化时创建的连接数 maxcached5, # 连接池中最多闲置的连接数 maxshared3, # 共享连接数 blockingTrue, # 连接池满时是否阻塞等待 **DB_CONFIG ) # 使用连接池 def get_user_with_pool(user_id): connection pool.connection() try: with connection.cursor() as cursor: sql SELECT * FROM users WHERE id %s cursor.execute(sql, (user_id,)) return cursor.fetchone() finally: connection.close()4.3 使用ORM框架对于大型项目可以考虑使用SQLAlchemy等ORM框架from sqlalchemy import create_engine, Column, Integer, String from sqlalchemy.ext.declarative import declarative_base from sqlalchemy.orm import sessionmaker # 创建引擎 engine create_engine(mysqlpymysql://username:passwordlocalhost/test_db) # 定义模型 Base declarative_base() class User(Base): __tablename__ users id Column(Integer, primary_keyTrue) username Column(String(50), uniqueTrue) email Column(String(100), uniqueTrue) password Column(String(100)) # 创建表 Base.metadata.create_all(engine) # 创建会话 Session sessionmaker(bindengine) session Session() # 使用ORM操作 new_user User(usernametest, emailtestexample.com, password123456) session.add(new_user) session.commit() # 查询 user session.query(User).filter_by(usernametest).first() print(user.email)5. 安全注意事项SQL注入防护永远不要直接拼接SQL语句一定要使用参数化查询# 错误做法有SQL注入风险 cursor.execute(SELECT * FROM users WHERE username username ) # 正确做法 cursor.execute(SELECT * FROM users WHERE username %s, (username,))密码安全不要在数据库中存储明文密码应该存储加盐哈希import hashlib import os def hash_password(password): salt os.urandom(32) # 随机盐值 key hashlib.pbkdf2_hmac(sha256, password.encode(utf-8), salt, 100000) return salt key连接安全生产环境不要使用root账户限制数据库用户的权限考虑使用SSL连接错误处理不要将数据库错误直接暴露给用户应该捕获并记录try: # 数据库操作 except pymysql.Error as e: logger.error(f数据库错误: {e}) raise DatabaseError(操作失败请稍后再试)6. 性能优化技巧批量操作尽可能使用批量插入、批量更新减少网络往返# 批量插入1000条数据 data [(fuser{i}, fuser{i}example.com) for i in range(1000)] cursor.executemany(INSERT INTO users (username, email) VALUES (%s, %s), data)索引优化为常用查询条件添加索引cursor.execute(CREATE INDEX idx_username ON users(username))连接复用在Web应用中可以考虑在请求开始时获取连接结束时释放而不是每个SQL操作都创建新连接。查询优化只查询需要的列避免SELECT *# 不好 cursor.execute(SELECT * FROM users WHERE id 1) # 更好 cursor.execute(SELECT username, email FROM users WHERE id 1)使用EXPLAIN分析慢查询cursor.execute(EXPLAIN SELECT * FROM users WHERE username LIKE a%) print(cursor.fetchall())7. 常见问题与解决方案连接超时现象pymysql.err.OperationalError: (2013, Lost connection to MySQL server during query)解决方案增加connect_timeout参数或检查MySQL的wait_timeout设置编码问题现象插入中文出现乱码解决方案确保连接时指定了正确的字符集推荐utf8mb4连接数过多现象pymysql.err.OperationalError: (1040, Too many connections)解决方案使用连接池或增加MySQL的max_connections死锁问题现象pymysql.err.OperationalError: (1213, Deadlock found when trying to get lock)解决方案优化事务大小和顺序添加适当的索引主键冲突现象pymysql.err.IntegrityError: (1062, Duplicate entry 1 for key PRIMARY)解决方案检查是否忘记设置自增主键或手动指定了重复ID在实际项目中我发现大多数数据库问题都可以通过以下方式避免始终使用参数化查询防止SQL注入每个数据库操作后立即检查返回值或捕获异常为生产环境配置适当的连接池为常用查询添加合适的索引定期检查慢查询日志并优化