Python连接MySQL数据库实战指南

引言

Python作为当今最流行的编程语言之一，与MySQL数据库的交互能力是其重要特性之一。在实际开发中，约75%的Python应用程序需要与数据库进行数据交换。这种组合在以下场景中尤为常见：

• Web开发（如Django、Flask框架）
• 数据分析与数据科学（Pandas数据处理）
• 自动化脚本（数据迁移、报表生成）
• 机器学习（特征存储、模型参数持久化）

主流Python MySQL连接库包括：

• mysql-connector-python（MySQL官方驱动）
• pymysql（纯Python实现，兼容性更好）
• MySQLdb（C扩展实现，性能较高）

准备工作

环境配置

安装连接库（以pymysql为例）：

pip install pymysql
# 或者安装官方驱动
pip install mysql-connector-python

MySQL服务准备

1. 确保MySQL服务正在运行：

   # Linux系统
   sudo service mysql status
   # Windows系统
   net start mysql
   

2. 创建测试用户并授权：

   CREATE USER 'testuser'@'localhost' IDENTIFIED BY 'password';
   GRANT ALL PRIVILEGES ON *.* TO 'testuser'@'localhost';
   FLUSH PRIVILEGES;
   

创建测试数据库

CREATE DATABASE python_test;
USE python_test;

CREATE TABLE users (
    id INT AUTO_INCREMENT PRIMARY KEY,
    name VARCHAR(50) NOT NULL,
    email VARCHAR(100) UNIQUE,
    created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP
);

连接MySQL数据库的基本方法

使用mysql-connector-python

import mysql.connector

config = {
    'host': 'localhost',
    'user': 'testuser',
    'password': 'password',
    'database': 'python_test',
    'charset': 'utf8mb4'  # 支持完整Unicode
}

try:
    conn = mysql.connector.connect(**config)
    print("连接成功！")
except mysql.connector.Error as err:
    print(f"连接失败: {err}")

使用pymysql

import pymysql

connection = pymysql.connect(
    host='localhost',
    user='testuser',
    password='password',
    database='python_test',
    cursorclass=pymysql.cursors.DictCursor  # 返回字典形式结果
)

连接参数详解

执行SQL查询

基本操作流程

1. 创建游标对象
2. 执行SQL语句
3. 获取结果（SELECT查询）
4. 提交事务（非自动提交时）
5. 关闭游标和连接

示例代码

# 查询操作
with connection.cursor() as cursor:
    cursor.execute("SELECT * FROM users WHERE name LIKE %s", ('张%',))
    results = cursor.fetchall()
    for row in results:
        print(row)

# 插入操作
try:
    with connection.cursor() as cursor:
        sql = "INSERT INTO users (name, email) VALUES (%s, %s)"
        cursor.execute(sql, ('张三', 'zhangsan@example.com'))
    connection.commit()  # 手动提交
except:
    connection.rollback()  # 发生错误时回滚

参数化查询的重要性

防止SQL注入的示例对比：

# 不安全的方式
name = "张三'; DROP TABLE users;--"
cursor.execute(f"SELECT * FROM users WHERE name = '{name}'")

# 安全的方式（参数化查询）
cursor.execute("SELECT * FROM users WHERE name = %s", (name,))

事务管理

事务控制示例

try:
    connection.begin()  # 开始事务
    with connection.cursor() as cursor:
        # 执行多个SQL语句
        cursor.execute("UPDATE accounts SET balance = balance - 100 WHERE user_id = 1")
        cursor.execute("UPDATE accounts SET balance = balance + 100 WHERE user_id = 2")
    connection.commit()  # 提交事务
except Exception as e:
    connection.rollback()  # 回滚事务
    print(f"事务失败: {e}")

自动提交设置

# 创建连接时设置
connection = pymysql.connect(..., autocommit=True)

# 或者后续修改
connection.autocommit(True)

使用ORM框架（SQLAlchemy示例）

基本配置

from sqlalchemy import create_engine, Column, Integer, String, DateTime
from sqlalchemy.ext.declarative import declarative_base
from sqlalchemy.orm import sessionmaker

# 连接字符串格式：dialect+driver://username:password@host:port/database
engine = create_engine('mysql+pymysql://testuser:password@localhost/python_test')
Base = declarative_base()

# 定义模型
class User(Base):
    __tablename__ = 'users'
    
    id = Column(Integer, primary_key=True)
    name = Column(String(50))
    email = Column(String(100))
    created_at = Column(DateTime)

基本操作

# 创建会话
Session = sessionmaker(bind=engine)
session = Session()

# 添加记录
new_user = User(name='李四', email='lisi@example.com')
session.add(new_user)
session.commit()

# 查询记录
users = session.query(User).filter(User.name.like('李%')).all()

性能优化技巧

连接池配置

from sqlalchemy.pool import QueuePool

engine = create_engine(
    'mysql+pymysql://testuser:password@localhost/python_test',
    poolclass=QueuePool,
    pool_size=5,
    max_overflow=10,
    pool_timeout=30
)

批量操作

# 批量插入
data = [{'name': f'用户{i}', 'email': f'user{i}@example.com'} for i in range(100)]
with connection.cursor() as cursor:
    cursor.executemany(
        "INSERT INTO users (name, email) VALUES (%(name)s, %(email)s)",
        data
    )
connection.commit()

查询优化建议

只查询需要的列

执行查询时，明确指定需要返回的列名，而不是使用通配符*。这样可以减少网络传输的数据量和内存消耗，提高查询效率。例如：

-- 不推荐
SELECT * FROM users;

-- 推荐
SELECT id, username, email FROM users;

合理使用索引

1. 为经常用于查询条件的列创建索引
2. 为常用于JOIN操作的列创建索引
3. 避免对索引列使用函数或计算
4. 复合索引遵循最左前缀原则

示例：

-- 创建索引
CREATE INDEX idx_user_email ON users(email);

-- 复合索引
CREATE INDEX idx_user_name_status ON users(last_name, first_name, status);

避免SELECT *操作

SELECT *会返回表中所有列，包括不需要的列，这会：

1. 增加I/O操作
2. 占用更多内存
3. 降低网络传输效率
4. 当表结构变更时可能导致应用层错误

使用LIMIT分页

对于大数据集查询，使用LIMIT进行分页处理：

-- 基本分页
SELECT id, name FROM products LIMIT 10 OFFSET 20;

-- 更高效的分页方式（MySQL）
SELECT id, name FROM products WHERE id > 1000 LIMIT 10;

复杂查询考虑使用存储过程

对于复杂的业务逻辑查询，可以考虑使用存储过程：

1. 减少网络传输
2. 预编译提高执行效率
3. 便于维护和复用

示例：

CREATE PROCEDURE GetUserOrders(IN userId INT)
BEGIN
    SELECT o.order_id, o.order_date, p.product_name, oi.quantity
    FROM orders o
    JOIN order_items oi ON o.order_id = oi.order_id
    JOIN products p ON oi.product_id = p.product_id
    WHERE o.user_id = userId;
END;

常见问题与解决方案

连接问题

• 连接超时：增加connect_timeout参数
• Can't connect to MySQL server：检查服务是否运行，防火墙设置
• Lost connection：使用连接池或重试机制

编码问题

• 统一使用utf8mb4字符集
• 连接字符串添加charset参数

权限问题

• 确保用户有足够权限
• 检查主机限制（如'user'@'%' vs 'user'@'localhost'）

总结

方法对比

适用场景分析

mysql-connector

• 需要与MySQL紧密集成的项目
• 对性能要求极高的应用
• 需要官方认证支持的商业项目
• 示例：高频交易系统、数据分析平台

pymysql

• 需要兼容旧版MySQLdb接口的项目
• 轻量级Web应用
• 需要快速上手的教学项目
• 示例：小型博客系统、个人项目

SQLAlchemy

• 需要数据库抽象层的复杂应用
• 多数据库支持的项目
• 需要使用ORM模式开发的项目
• 示例：企业级ERP系统、SaaS平台

性能优化建议

1. mysql-connector优化：

   • 使用预编译语句(prepared statements)
   • 合理设置fetch_size参数
   • 启用连接压缩协议

2. pymysql优化：

   • 使用连接池管理连接
   • 批量操作时使用executemany
   • 避免频繁建立/关闭连接

3. SQLAlchemy优化：

   • 合理配置连接池大小
   • 使用Core API处理高性能需求
   • 启用批量操作(bulk_insert等)

最佳实践

1. 始终使用参数化查询

• 安全优势：防止SQL注入攻击，当用户输入包含恶意SQL片段时能确保安全
• 性能优势：数据库可以缓存编译后的查询计划
• 示例：

  # 不安全的拼接方式
  query = "SELECT * FROM users WHERE username = '" + username + "'"
  
  # 安全的参数化查询
  query = "SELECT * FROM users WHERE username = %s"
  cursor.execute(query, (username,))
  

2. 合理管理连接

• 使用with语句：确保连接在使用后自动关闭

  with pymysql.connect(...) as conn:
      with conn.cursor() as cursor:
          cursor.execute("SELECT * FROM users")
  

• 连接池技术：在高并发场景下提升性能
  • 常用连接池：DBUtils、SQLAlchemy的池功能
  • 配置参数：最大连接数、超时时间、回收策略

3. 生产环境使用ORM框架

• 推荐框架：
  • SQLAlchemy（功能全面）
  • Django ORM（Django项目集成）
  • Peewee（轻量级）
• 优势：
  • 抽象底层SQL，提高开发效率
  • 内置安全机制
  • 支持数据库迁移
  • 跨数据库兼容性

4. 重要操作添加事务支持

• ACID原则：确保原子性、一致性、隔离性、持久性
• 典型应用场景：
  • 银行转账（涉及多个账户余额更新）
  • 订单创建（同时更新库存和创建订单记录）
• 实现方式：

  try:
      conn.begin()
      # 执行多个SQL操作
      conn.commit()
  except Exception as e:
      conn.rollback()
      raise e
  

5. 添加适当的错误处理和日志记录

• 常见错误类型：
  • 连接错误（超时、认证失败）
  • 语法错误
  • 约束违反（唯一键、外键）
  • 超时和死锁
• 日志记录要点：
  • 记录错误堆栈
  • 敏感信息脱敏
  • 关键操作审计日志

延伸学习

1. MySQL官方文档

• 重点章节：
  • 性能优化（索引、查询优化）
  • 存储引擎比较（InnoDB vs MyISAM）
  • 复制与高可用方案
• 实用资源：
  • MySQL 8.0 Reference Manual
  • 官方博客的性能优化案例

2. SQLAlchemy官方教程

• 核心内容：
  • 核心API（Engine、Connection、MetaData）
  • ORM映射（Declarative Base）
  • 会话管理（Session的生命周期）
• 高级特性：
  • 关联关系（一对多、多对多）
  • 混合属性（Hybrid Attributes）
  • 事件监听系统

3. Python DB-API 2.0规范

• 关键接口：
  • connect() - 建立连接
  • cursor() - 创建游标
  • execute() - 执行查询
  • fetchone()/fetchall() - 获取结果
• 异常体系：
  • Warning - 非致命问题
  • Error - 所有错误基类
  • InterfaceError - 接口错误
  • DatabaseError - 数据库错误

4. 数据库设计原理

• 设计流程：
  1. 需求分析
  2. 概念设计（ER图）
  3. 逻辑设计（关系模型）
  4. 物理设计（索引、分区）
• 规范化理论：
  • 1NF（消除重复组）
  • 2NF（消除部分依赖）
  • 3NF（消除传递依赖）
  • BCNF（更强的3NF）
• 反规范化技巧：在特定场景下为提高性能而有意违反规范化原则