【超详细教程】Python 连接 MySQL 全流程实战

Ne0inhk

— 7 min read

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 分钟搞定）
  1. 安装 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;。
  2. 测试环境：
    • Python 版本：python --version（≥3.8）。
    • MySQL 版本：mysql --version（≥8.0，支持 JSON 等新特性）。
    • 防火墙/权限：确保 Python 主机能访问 MySQL（默认 localhost:3306）。

安装 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. 性能优化 & 最佳实践
优化点做法示例收益参考
连接复用用连接池，避免每次操作新 connect10-50x 加速
批量操作executemany() 而非循环 execute5-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/delete

main.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）？告诉我，我继续展开！

Read more

【算法通关指南：算法基础篇】二分算法：1.在排序树组中查找元素的第一个和最后一个位置 2.牛可乐和魔法封印

【算法通关指南：算法基础篇】二分算法：1.在排序树组中查找元素的第一个和最后一个位置 2.牛可乐和魔法封印

🔥小龙报：个人主页 🎬作者简介：C++研发，嵌入式，机器人方向学习者 ❄️个人专栏：《算法通关指南》 ✨ 永远相信美好的事情即将发生 文章目录 * 前言 * 一、二分算法 * 二、在排序树组中查找元素的第一个和最后一个位置 * 2.1题目 * 2.2 算法原理 * 2.3代码 * 三、牛可乐和魔法封印 * 3.1题目 * 3.2 算法原理 * 3.3代码 * 总结与每日励志 前言 本专栏聚焦算法题实战，系统讲解算法模块：以《c++编程》，《数据结构和算法》《基础算法》《算法实战》 等几个板块以题带点，讲解思路与代码实现，帮助大家快速提升代码能力ps:本章节题目分两部分，比较基础笔者只附上代码供大家参考，其他的笔者会附上自己的思考和讲解，希望和大家一起努力见证自己的算法成长 一、

By Ne0inhk
数据结构【栈和队列附顺序表应用算法】

数据结构【栈和队列附顺序表应用算法】

栈和队列和顺序表应用算法练习 * 1.栈 * 1.1概念与结构 * 1.2栈的实现 * 2.队列 * 2.1概念与结构 * 2.2队列的实现 * 3.附（顺序表应用算法） * 3.1移除元素 * 3.2删除有序数组中的重复项 * 3.3合并两个有序数组 1.栈 1.1概念与结构 栈：⼀种特殊的线性表，其只允许在固定的⼀端进行插入和删除元素操作。进行数据插入和删除操作的一端称为栈顶，另⼀端称为栈底。栈中的数据元素遵守后进先出的原则。 压栈：栈的插入操作叫做进栈/压栈/入栈，入数据在栈顶。 出栈：栈的删除操作叫做出栈，出数据也在栈顶。 1.2栈的实现 typedefint STDataType;typedefstructStack{ STDataType * a;int

By Ne0inhk
【笔试】算法的暴力美学——牛客 NC221681：dd爱框框

【笔试】算法的暴力美学——牛客 NC221681：dd爱框框

一、题目描述 二、算法原理 思路：滑动窗口 1）定义两个指针，一开始都为0，cur 从左开始遍历，定义一个 sum 来表示 prev 到 cur 的之间的值的总和，当 sum >= x 时，我们要根据题目条件来保存 prev 和 cur 的值； 2）当 sum >= x 时，我们记录完 prev 和 cur 的值的之后，sum -= arr[ prev ]，prev++ ，往后走，只要满足条件 sum >= x 我们就要记录

By Ne0inhk
python基于大数据的手机商品数据分析

python基于大数据的手机商品数据分析

目录 * 手机商品数据分析的摘要 * 开发技术路线 * 相关技术介绍 * 核心代码参考示例 * 结论 * 源码lw获取/同行可拿货,招校园代理 ：文章底部获取博主联系方式！ 手机商品数据分析的摘要 随着智能手机市场的快速发展，海量数据为产品优化和商业决策提供了重要依据。Python凭借其强大的数据处理和可视化能力，成为手机商品数据分析的核心工具。 数据采集阶段，可通过爬虫技术（如Scrapy、BeautifulSoup）从电商平台（京东、淘宝）获取手机销售数据，包括价格、销量、用户评价等。结构化数据存储于MySQL或MongoDB，非结构化数据（如评论文本）通过NLTK或jieba进行分词和情感分析。 数据清洗环节使用Pandas处理缺失值和异常值，例如剔除价格异常记录或填充平均评分。特征工程中，提取关键指标如品牌市场份额、价格分布、用户满意度，并利用Matplotlib或Seaborn生成可视化图表（如热力图展示品牌-价格相关性）。 机器学习模型（如随机森林、XGBoost）可预测销量趋势或用户购买意愿，特征重要性分析揭示影响销量的核心因素（如CP

By Ne0inhk