Python连接MySQL数据库完全指南
一、环境准备四部曲
1. 安装MySQL服务器（Docker极简版）
docker run --name mysql2025 -e MYSQL_ROOT_PASSWORD=123456 -p 3306:3306 -d mysql:8.4
2. 安装Python连接驱动
# 官方推荐量子加密版
pip install mysql-connector-python==8.3.0
# 经典兼容版（适合老系统）
pip install pymysql
3. 创建测试数据库
CREATE DATABASE school_db;
USE school_db;CREATE TABLE students (id INT AUTO_INCREMENT PRIMARY KEY,name VARCHAR(50),age INT,major VARCHAR(50)
);
二、连接数据库六步法
import mysql.connector
from mysql.connector import Error def create_connection():try:conn = mysql.connector.connect(host='localhost',user='root',password='123456',database='school_db',auth_plugin='caching_sha2_password' # MySQL 8.0+必填 )print("✅ 连接成功！服务器版本:", conn.get_server_info())return conn except Error as e:print(f"❌ 连接失败: {e}")return None
# 测试连接
connection = create_connection()
⚠️ 常见问题排查：
若报错Authentication plugin ‘caching_sha2_password’ → 改用mysql_native_password认证
连接超时可添加参数：connect_timeout=30
三、数据操作全流程（CRUD）
1. 数据插入（Create）
def insert_student(name, age, major):cursor = connection.cursor()sql = "INSERT INTO students (name, age, major) VALUES (%s, %s, %s)"val = (name, age, major)try:cursor.execute(sql, val)connection.commit()print(f"🆕 插入成功，ID: {cursor.lastrowid}")except Error as e:print(f"插入失败: {e}")insert_student("张伟", 20, "人工智能")
2. 数据查询（Read）
def query_students():cursor = connection.cursor(dictionary=True) # 返回字典格式 cursor.execute("SELECT * FROM students")print("📊 学生列表：")for row in cursor.fetchall():print(f"ID:{row['id']} | 姓名:{row['name']} | 专业:{row['major']}")query_students()
3. 数据更新（Update）
def update_student(student_id, new_major):cursor = connection.cursor()sql = "UPDATE students SET major = %s WHERE id = %s"cursor.execute(sql, (new_major, student_id))connection.commit()print(f"🔄 受影响行数: {cursor.rowcount}")update_student(1, "量子计算")
4. 数据删除（Delete）
def delete_student(student_id):cursor = connection.cursor()cursor.execute("DELETE FROM students WHERE id = %s", (student_id,))connection.commit()print(f"🗑️ 删除成功" if cursor.rowcount else "⚠️ 未找到该学生")delete_student(1)
四、2025安全规范
1. 连接池管理
from mysql.connector import pooling dbconfig = {"host":"localhost","user":"root","password":"123456","database":"school_db"
}pool = pooling.MySQLConnectionPool(pool_name="mypool",pool_size=5,**dbconfig
)# 从连接池获取连接
connection = pool.get_connection()
2. 参数化查询规范
# 危险做法 ❌
f"SELECT * FROM users WHERE name = '{name}'" # 安全做法 ✅
"SELECT * FROM users WHERE name = %s", (name,)
)