4 种 Python 连接 MySQL 数据库的方法

作者： 分类：所有,mysql 字数：9364

4 种 Python 连接 MySQL 数据库的方法

一、背景与问题

在现代软件开发中，数据库连接是核心能力之一。Python 作为通用编程语言，提供了多种连接 MySQL 的方式。然而，开发者常面临以下问题：

  • 如何选择适合不同场景的连接方式
  • 如何避免 SQL 注入等安全风险
  • 如何在高并发场景下优化性能
  • 如何处理连接池和事务管理
  • 如何在不同开发阶段（如开发、测试、生产）配置连接参数

本文将深入分析四种常见实现方式，结合真实开发场景，探讨其原理、适用场景、常见陷阱和优化策略。

二、基本原理

MySQL 是基于 TCP/IP 协议的客户端-服务器架构数据库。Python 连接 MySQL 的本质是通过网络协议与 MySQL 服务器建立通信链路，发送 SQL 查询语句并接收结果。

核心过程包含以下步骤：

  1. 建立 TCP 连接
  2. 发送认证信息（用户名、密码）
  3. 执行 SQL 语句
  4. 处理查询结果
  5. 关闭连接

不同连接方式在实现细节上存在差异，例如直接使用底层库（如 mysql-connector）与 ORM 框架（如 SQLAlchemy）在 SQL 转换、连接管理、异常处理等方面有显著区别。

三、环境准备

# 安装依赖
pip install mysql-connector-python pymysql sqlalchemy

需要确保 MySQL 服务已启动，并创建测试数据库和表：

CREATE DATABASE test_db;
USE test_db;

CREATE TABLE users (
    id INT AUTO_INCREMENT PRIMARY KEY,
    name VARCHAR(50),
    email VARCHAR(100)
);

INSERT INTO users (name, email) VALUES ('Alice', 'alice@example.com'), ('Bob', 'bob@example.com');

四、核心实现

方法一：使用 mysql-connector（官方库）

import mysql.connector
from mysql.connector import Error

def connect_with_connector():
    try:
        connection = mysql.connector.connect(
            host='localhost',
            database='test_db',
            user='root',
            password='password'
        )
        if connection.is_connected():
            cursor = connection.cursor()
            cursor.execute("SELECT * FROM users")
            rows = cursor.fetchall()
            for row in rows:
                print(row)
    except Error as e:
        print(f"Error: {e}")
    finally:
        if 'connection' in locals() and connection.is_connected():
            cursor.close()
            connection.close()
            print("MySQL connection is closed")

关键代码解释：

  1. mysql.connector.connect 建立 TCP 连接
  2. cursor.execute() 将 SQL 语句发送到服务器
  3. fetchall() 获取结果集
  4. 使用 try...finally 确保连接关闭
  5. is_connected() 检查连接状态

适用场景：

  • 需要直接操作底层 API
  • 对性能敏感的场景（如批量处理）
  • 需要精细控制事务的场景

注意事项：

  • 不推荐用于生产环境，缺乏 ORM 层
  • 需要处理连接池和超时问题

方法二：使用 pymysql（第三方库）

import pymysql

def connect_with_pymysql():
    connection = pymysql.connect(
        host='localhost',
        user='root',
        password='password',
        db='test_db',
        charset='utf8mb4',
        cursorclass=pymysql.cursors.DictCursor
    )
    try:
        with connection.cursor() as cursor:
            sql = "SELECT * FROM users"
            cursor.execute(sql)
            results = cursor.fetchall()
            for row in results:
                print(row)
    finally:
        connection.close()

关键代码解释：

  1. pymysql.connect 建立连接，支持上下文管理器
  2. DictCursor 返回字典形式的结果
  3. 使用 with 语句自动管理游标生命周期
  4. 更好的异常处理和连接管理

性能优化：

  • 使用 cursor.execute() 批量执行
  • 启用 use_unicode=True 支持中文
  • 使用连接池（如 pymysqlpool）处理高并发

安全风险：

  • 需要避免 SQL 注入，使用参数化查询：

    sql = "SELECT * FROM users WHERE email = %s"
cursor.execute(sql, (email,))

方法三：使用 SQLAlchemy ORM（高级抽象）

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

Base = declarative_base()

class User(Base):
    __tablename__ = 'users'
    id = Column(Integer, primary_key=True)
    name = Column(String(50))
    email = Column(String(100))

engine = create_engine('mysql+pymysql://root:password@localhost/test_db')
Session = sessionmaker(bind=engine)

def connect_with_sqlalchemy():
    session = Session()
    try:
        users = session.query(User).all()
        for user in users:
            print(f"{user.name} - {user.email}")
    finally:
        session.close()

关键代码解释：

  1. 使用 SQLAlchemy 的 ORM 层抽象 SQL
  2. 自动处理连接池和事务
  3. 通过类定义映射数据库表结构
  4. 使用 session 管理数据库会话

性能考量：

  • ORM 层引入额外开销（约 10-30% 性能损耗）
  • 需要合理使用 query 和 session 管理
  • 支持异步 ORM（sqlalchemy-async）

适用场景：

  • 快速开发场景（减少 SQL 编写）
  • 需要跨平台数据迁移
  • 需要代码级 ORM 约束（如外键、唯一性）

五、完整案例：学生信息管理系统

# student_manager.py
import mysql.connector
from mysql.connector import Error

def create_table():
    try:
        connection = mysql.connector.connect(
            host='localhost',
            user='root',
            password='password',
            database='test_db'
        )
        cursor = connection.cursor()
        cursor.execute("""
            CREATE TABLE IF NOT EXISTS students (
                id INT AUTO_INCREMENT PRIMARY KEY,
                name VARCHAR(100),
                email VARCHAR(100) UNIQUE
            )
        """)
    except Error as e:
        print(f"Error creating table: {e}")
    finally:
        if connection.is_connected():
            cursor.close()
            connection.close()

def add_student(name, email):
    try:
        connection = mysql.connector.connect(
            host='localhost',
            user='root',
            password='password',
            database='test_db'
        )
        cursor = connection.cursor()
        sql = "INSERT INTO students (name, email) VALUES (%s, %s)"
        cursor.execute(sql, (name, email))
        connection.commit()
        print("Student added successfully")
    except Error as e:
        print(f"Error: {e}")
    finally:
        if connection.is_connected():
            cursor.close()
            connection.close()

def list_students():
    try:
        connection = mysql.connector.connect(
            host='localhost',
            user='root',
            password='password',
            database='test_db'
        )
        cursor = connection.cursor()
        cursor.execute("SELECT * FROM students")
        for row in cursor.fetchall():
            print(row)
    except Error as e:
        print(f"Error: {e}")
    finally:
        if connection.is_connected():
            cursor.close()
            connection.close()

# 使用示例
if __name__ == "__main__":
    create_table()
    add_student("Alice", "alice@example.com")
    list_students()

运行流程：

  1. 创建学生表
  2. 添加学生记录
  3. 查询并打印所有学生

关键改进点：

  • 使用参数化查询防止 SQL 注入
  • 分离创建表和操作数据的逻辑
  • 添加异常处理确保资源释放

六、源码解析

以 pymysql 的连接池实现为例：

from pymysql import pool

# 创建连接池
pool = pool.Pool(
    host='localhost',
    user='root',
    password='password',
    db='test_db',
    size=10  # 最大连接数
)

# 获取连接
conn = pool.get_conn()
cursor = conn.cursor()
cursor.execute("SELECT * FROM users")
results = cursor.fetchall()
cursor.close()
pool.put_conn(conn)

核心机制：

  1. 连接池预先创建多个连接
  2. 线程安全的连接管理
  3. 避免频繁创建/销毁连接的开销

性能优化：

  • 设置合理 size 防止资源浪费
  • 使用 thread_local 管理连接
  • 配合 keepalive 参数维持空闲连接

七、进阶使用

1. 异步连接（使用 asyncmy）

import asyncio
from asyncmy import connect

async def async_query():
    async with await connect('mysql+pymysql://root:password@localhost/test_db') as conn:
        async with await conn.cursor() as cur:
            await cur.execute("SELECT * FROM users")
            results = await cur.fetchall()
            print(results)

适用场景：

  • 高并发 I/O 密集型应用
  • 异步框架（如 FastAPI、Tornado）

2. 使用连接池（pymysqlpool）

from pymysqlpool import Pool

pool = Pool(
    host='localhost',
    user='root',
    password='password',
    database='test_db',
    max_connections=10
)

conn = pool.get_connection()
cursor = conn.cursor()
cursor.execute("SELECT * FROM users")

优势：

  • 自动管理连接生命周期
  • 支持连接健康检查

八、性能与工程实践

1. 性能优化策略

方法优化点效果
使用连接池减少连接创建开销提升 30% 吞吐量
批量操作减少网络往返降低 50% 延迟
索引优化加速查询提升 2-10 倍速度
避免 SELECT *减少数据传输降低 30% 网络开销

2. 异常处理建议

try:
    with connection.cursor() as cursor:
        cursor.execute("SELECT * FROM non_existent_table")
except mysql.connector.ProgrammingError as e:
    print(f"Query error: {e}")

3. 安全实践

  • 使用 parameterized 查询
  • 设置 sql_mode=ONLY_FULL_GROUP_BY
  • 配置 MySQL 的 query_cache_size 为 0
  • 限制数据库用户权限（最小权限原则）

九、常见问题与踩坑

1. 网络问题

错误示例：

connection = mysql.connector.connect(host='127.0.0.1')  # 错误：未指定端口和数据库

正确方式：

connection = mysql.connector.connect(
    host='127.0.0.1',
    port=3306,
    database='test_db',
    user='root',
    password='password'
)

2. 索引问题

错误示例：

SELECT * FROM users WHERE name LIKE '%Alice%'

优化建议：

  • 建立 name 字段的索引
  • 使用 LIKE 'Alice%' 前缀查询
  • 避免 SELECT *，减少 I/O

3. 配置问题

常见错误：

  • 未设置 use_unicode=True 导致中文乱码
  • 未配置 charset='utf8mb4' 支持 emoji
  • 未设置 connect_timeout 导致连接超时

解决方案：

connection = mysql.connector.connect(
    host='localhost',
    user='root',
    password='password',
    database='test_db',
    connect_timeout=5,
    charset='utf8mb4'
)

十、最佳实践

1. 建议使用方案

场景推荐方式说明
快速开发SQLAlchemy ORM简化 SQL 编写
高性能场景pymysql + 连接池原生控制
异步系统asyncmy + FastAPI非阻塞 I/O
安全敏感参数化查询 + 检查点防止 SQL 注入

2. 避免使用方案

场景不推荐方式原因
生产环境mysql-connector缺乏 ORM 支持
高并发无连接池资源浪费
安全敏感SQL 拼接高危漏洞
跨平台硬编码连接参数配置管理困难

十一、总结

Python 连接 MySQL 的方式多种多样，每种方法都有其适用场景和优缺点。选择合适的方式需要考虑以下因素：

  • 开发阶段：快速开发 vs 性能敏感
  • 项目规模：小型项目 vs 大型系统
  • 安全需求：是否需要防注入
  • 异常处理：是否需要精细控制
  • 系统架构：是否需要异步支持

在实际开发中，建议遵循以下原则：

  1. 开发阶段使用 ORM，提高开发效率
  2. 生产环境使用连接池，优化资源利用率
  3. 所有查询使用参数化，杜绝 SQL 注入
  4. 定期进行性能调优，包括索引、查询、连接池等
  5. 配置管理分离，避免硬编码数据库参数

通过合理选择连接方式，结合性能优化和安全实践，可以构建出高效、稳定、安全的数据库系统。

评论已关闭

推荐阅读

AIGC实战——Transformer模型
2024年12月01日
python , AI , 机器学习
Socket TCP 和 UDP 编程基础（Python）
2024年11月30日
python , tcp , udp
如何使用 ChatGPT 进行学术润色？你需要这些指令
2024年12月01日
AI
最新 Python 调用 OpenAi 详细教程实现问答、图像合成、图像理解、语音合成、语音识别（详细教程）
2024年11月24日
python , AI
ChatGPT 和 DALL·E 2 配合生成故事绘本
2024年12月01日
python , AI
omegaconf，一个超强的 Python 库！
2024年11月24日
python
【视觉AIGC识别】误差特征、人脸伪造检测、其他类型假图检测
2024年12月01日
python , AI
［超级详细］如何在深度学习训练模型过程中使用 GPU 加速
2024年11月29日
python , AI , 机器学习
Python 物理引擎pymunk最完整教程
2024年11月27日
python
MediaPipe 人体姿态与手指关键点检测教程
2024年11月27日
python
深入了解 Taipy：Python 打造 Web 应用的全面教程
2024年11月26日
python
基于Transformer的时间序列预测模型
2024年11月25日
python , Matplotlib , 机器学习
Python在金融大数据分析中的AI应用（股价分析、量化交易）实战
2024年11月25日
python , Matplotlib , 机器学习
AIGC Gradio系列学习教程之Components
2024年12月01日
python , AI
Python3 `asyncio` — 异步 I/O，事件循环和并发工具
2024年11月30日
python
llama-factory SFT系列教程：大模型在自定义数据集 LoRA 训练与部署
2024年12月01日
python , AI , 机器学习
Python 多线程和多进程用法
2024年11月24日
python
Python socket详解，全网最全教程
2024年11月27日
python , http , websocket
python之plot()和subplot()画图
2024年11月26日
python , Matplotlib
理解 DALL·E 2、Stable Diffusion 和 Midjourney 工作原理
2024年12月01日
python , AI , 机器学习
Copyright © 2026 Tech Blog
All Right Reserved | 京ICP备19053165号