【超详细教程】Python 连接 MySQL 全流程实战
Python 连接 MySQL 全流程实战教程（2026版超详细指南）
以下是基于 Python 3.12+ 和 MySQL 8.0+ 的从零到精通实战教程。我会按实际开发流程组织内容：环境准备 → 连接基础 → CRUD 操作 → 高级技巧 → 性能优化 → 完整项目示例 → 常见问题排查。
这个教程假设你有基本的 Python 知识（如变量、函数、异常处理）。我们优先使用 mysql-connector-python（Oracle 官方驱动，推荐），备用 pymysql（纯 Python，轻量）。
为什么学这个？
- MySQL 是最流行的关系型数据库（2026 年市场份额仍超 40%）。
- Python + MySQL 是 Web/App/数据分析的黄金组合（e.g., Django/Flask 后端、数据 ETL）。
- 掌握后，能轻松扩展到其他数据库（如 PostgreSQL via psycopg2）。
0. 环境准备（5 分钟搞定）
- 安装 MySQL 服务器（如果没有）：
- Windows/Mac：下载 MySQL Community Server（官网 mysql.com）。
- Linux：
sudo apt install mysql-server(Ubuntu) 或sudo yum install mysql-server(CentOS)。 - 启动：
mysql -u root -p测试连接（默认密码为空或 root）。 - 创建测试数据库：登录 MySQL 后，执行
CREATE DATABASE test_db CHARACTER SET utf8mb4;。
- 测试环境：
- Python 版本：
python --version（≥3.8）。 - MySQL 版本：
mysql --version（≥8.0，支持 JSON 等新特性）。 - 防火墙/权限：确保 Python 主机能访问 MySQL（默认 localhost:3306）。
- Python 版本：
安装 Python 驱动（必须）：
# 推荐官方驱动（支持连接池、预编译） pip install mysql-connector-python==9.0.0 # 2026 最新稳定版# 备用：纯 Python 驱动（无 C 依赖，轻便） pip installpymysql==1.1.1 1. 基础连接（Hello World 级）
核心：用 connect() 创建连接对象。
import mysql.connector # 或 import pymysql# 配置（替换为你自己的） config ={'host':'localhost',# 服务器地址（云数据库用 IP/域名）'port':3306,# 默认端口'user':'root',# 用户名'password':'your_password',# 密码'database':'test_db',# 数据库名'charset':'utf8mb4',# 支持 emoji 等'auth_plugin':'mysql_native_password'# 兼容旧客户端}# 建立连接try: conn = mysql.connector.connect(**config)# pymysql 用 pymysql.connect(**config)print("连接成功！MySQL 版本：", conn.get_server_info())except mysql.connector.Error as err:print(f"连接失败：{err}")finally:if'conn'inlocals()and conn.is_connected(): conn.close()print("连接已关闭。")输出示例：
连接成功！MySQL 版本： 8.0.35 连接已关闭。 关键点：
- 异常处理：总是用 try-except（常见错误：权限、密码、网络）。
- 关闭连接：防止资源泄漏（用 with 语句更好，后面讲）。
- pymysql 差异：几乎相同，但 pymysql 是线程安全的（多线程首选）。
2. CRUD 操作全解（Create/Read/Update/Delete）
用 游标（Cursor） 执行 SQL。
- 步骤：连接 → 创建游标 → 执行 SQL → 处理结果 → 提交/回滚 → 关闭。
2.1 Create（插入数据）
先建表（只执行一次）：
# 建表 SQL（示例：用户表） create_table_sql =""" CREATE TABLE IF NOT EXISTS users ( id INT AUTO_INCREMENT PRIMARY KEY, name VARCHAR(50) NOT NULL, age INT, email VARCHAR(100) UNIQUE, created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP ) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4; """ conn = mysql.connector.connect(**config) cursor = conn.cursor() cursor.execute(create_table_sql) conn.commit()# DDL 也需 commit cursor.close() conn.close()插入单条/多条：
conn = mysql.connector.connect(**config) cursor = conn.cursor()# 单条插入 insert_sql ="INSERT INTO users (name, age, email) VALUES (%s, %s, %s)" data =("Alice",28,"[email protected]") cursor.execute(insert_sql, data)# 多条插入（高效） multi_data =[("Bob",35,"[email protected]"),("Charlie",22,"[email protected]")] cursor.executemany(insert_sql, multi_data) conn.commit()# 必须 commit，否则数据不保存print(f"插入 {cursor.rowcount} 条记录。") cursor.close() conn.close()安全提示：用 %s 占位符防 SQL 注入（预编译）。
2.2 Read（查询数据）
conn = mysql.connector.connect(**config) cursor = conn.cursor(dictionary=True)# 返回 dict，便于访问 (e.g., row['name'])# 查询所有 select_all_sql ="SELECT * FROM users ORDER BY age DESC" cursor.execute(select_all_sql) rows = cursor.fetchall()# 全部结果for row in rows:print(row)# e.g., {'id': 1, 'name': 'Alice', ...}# 查询单条/带条件 select_one_sql ="SELECT name, age FROM users WHERE id = %s" cursor.execute(select_one_sql,(1,)) row = cursor.fetchone()# 第一条print("单条：", row)# 分页（LIMIT + OFFSET） paged_sql ="SELECT * FROM users LIMIT %s OFFSET %s" cursor.execute(paged_sql,(5,0))# 前5条，从0开始 rows = cursor.fetchall() cursor.close() conn.close()输出示例：
{'id': 2, 'name': 'Bob', 'age': 35, 'email': '[email protected]', 'created_at': datetime.datetime(2026, 1, 21, 18, 0, 0)} 单条： {'name': 'Alice', 'age': 28} 进阶：用 buffered=True 游标处理大结果集。
2.3 Update（更新数据）
conn = mysql.connector.connect(**config) cursor = conn.cursor() update_sql ="UPDATE users SET age = %s WHERE name = %s" cursor.execute(update_sql,(30,"Alice")) conn.commit()print(f"更新 {cursor.rowcount} 条记录。") cursor.close() conn.close()2.4 Delete（删除数据）
conn = mysql.connector.connect(**config) cursor = conn.cursor() delete_sql ="DELETE FROM users WHERE age < %s" cursor.execute(delete_sql,(25,)) conn.commit()print(f"删除 {cursor.rowcount} 条记录。") cursor.close() conn.close()事务管理：多操作用 conn.start_transaction() 和 conn.rollback()（出错回滚）。
3. 用 with 语句管理连接/游标（推荐实践）
简化 finally/close 逻辑。
import mysql.connector with mysql.connector.connect(**config)as conn:with conn.cursor()as cursor: cursor.execute("SELECT COUNT(*) FROM users") count = cursor.fetchone()[0]print(f"总记录数：{count}")# with 结束自动 commit/close# conn 也自动关闭4. 高级技巧（生产级应用）
4.1 连接池（高并发必备）
mysql-connector 支持内置池。
from mysql.connector import pooling pool = pooling.MySQLConnectionPool( pool_name="mypool", pool_size=10,# 最大连接数**config )# 获取连接 conn = pool.get_connection() cursor = conn.cursor() cursor.execute("SELECT * FROM users LIMIT 1")print(cursor.fetchone()) cursor.close() conn.close()# 归还到池优势：复用连接，减少开销（Web App 首选）。
4.2 存储过程/批量操作
调用存储过程：
cursor.callproc("my_stored_proc",(arg1, arg2))for result in cursor.stored_results():print(result.fetchall())批量插入（10万+ 数据）：
用 executemany() + 事务。
4.3 Pandas 集成（数据分析神器）
import pandas as pd import mysql.connector conn = mysql.connector.connect(**config) df = pd.read_sql("SELECT * FROM users WHERE age > 25", con=conn)print(df.head())# 写回数据库 df.to_sql("users_backup", con=conn, if_exists="replace", index=False) conn.close()2026 提示：Pandas 3.0+ 支持 Arrow backend，读写更快。
4.4 异步连接（asyncio，高吞吐）
用 aiomysql（基于 pymysql）。
pip install aiomysql import asyncio import aiomysql asyncdefmain(): pool =await aiomysql.create_pool(**config)asyncwith pool.acquire()as conn:asyncwith conn.cursor()as cur:await cur.execute("SELECT * FROM users") rows =await cur.fetchall()print(rows) pool.close()await pool.wait_closed() asyncio.run(main())5. 性能优化 & 最佳实践
| 优化点 | 做法示例 | 收益参考 |
|---|---|---|
| 连接复用 | 用连接池，避免每次操作新 connect | 10-50x 加速 |
| 批量操作 | executemany() 而非循环 execute | 5-20x 快 |
| 索引利用 | SQL 加 WHERE id=xx（id 有索引） | 查询 100x+ |
| 预编译语句 | cursor.prepare(sql) 复用 | 安全 + 快 |
| 读写分离 | 主库写，从库读（config 切换 host） | 高可用 |
| 错误重试 | 用 retry 库自动重连（pip install retry） | 鲁棒性 ↑ |
- 监控：用
conn.cmd_query("SHOW STATUS LIKE 'Threads%'")查连接数。 - 安全：用环境变量存密码（os.getenv(“DB_PASS”)）。
- 日志：加
conn.autocommit = False手动控制。
6. 完整项目实战：简单用户管理系统
项目结构：
user_manager/ ├── config.py # 配置 ├── db.py # 数据库操作 └── main.py # 入口 config.py：
DB_CONFIG ={'host':'localhost','user':'root','password':'your_password','database':'test_db','charset':'utf8mb4'}db.py：
import mysql.connector from config import DB_CONFIG defget_connection():return mysql.connector.connect(**DB_CONFIG)defadd_user(name, age, email):with get_connection()as conn:with conn.cursor()as cursor: sql ="INSERT INTO users (name, age, email) VALUES (%s, %s, %s)" cursor.execute(sql,(name, age, email)) conn.commit()defget_users():with get_connection()as conn:with conn.cursor(dictionary=True)as cursor: cursor.execute("SELECT * FROM users")return cursor.fetchall()# 类似地加 update/deletemain.py：
from db import add_user, get_users add_user("David",40,"[email protected]") users = get_users()for user in users:print(user)运行：python main.py
输出：所有用户列表。
扩展：加 Flask API（/users GET/POST）变成 Web 服务。
7. 常见问题排查（Top 10）
| 错误代码/描述 | 原因 & 解决 |
|---|---|
| Access denied (1045) | 密码/用户错 → 检查 config |
| Can’t connect (2003) | 网络/端口 → ping host, 查防火墙 |
| Unknown database (1049) | 数据库不存在 → CREATE DATABASE |
| SQL syntax error (1064) | SQL 写错 → 用工具测试 SQL |
| Lost connection (2013) | 超时 → 加 ‘connection_timeout’: 300 |
| Too many connections (1040) | 连接过多 → 用池，加大 max_connections |
| UnicodeEncodeError | 编码问题 → 加 charset=‘utf8mb4’ |
| OperationalError: 2055 | 缓存 SHA2 → 改 auth_plugin |
| pymysql vs mysql-connector | 选 pymysql 如果无 C 扩展需求 |
| Slow query | 加索引：ALTER TABLE ADD INDEX (col) |
调试技巧：
- 打印 err.errno / err.msg。
- 用 MySQL Workbench 测试 SQL。
- 日志：加
mysql.connector.logging配置。
恭喜！你现在能独立用 Python 操作 MySQL 了～
有具体部分想深入？如 ORM（SQLAlchemy）、云数据库（AWS RDS）、大数据集成（PySpark + MySQL）？告诉我，我继续展开！
