Python连接MySQL数据库完全指南

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,)