🐍 Python Day47：Python 操作 MySQL — 生产级数据库

🕐 预计用时：2-3 小时 ｜ 🎯 目标：掌握 pymysql 连接配置、增删改查、防 SQL 注入

📖 今日目录

1. MySQL 简介与安装
2. pymysql 安装与连接
3. 创建数据库和表
4. 增删改查 (CRUD)
5. 参数化查询 — 防 SQL 注入
6. 事务管理
7. 连接池
8. 错误处理
9. 实战：用户管理系统
10. SQLite vs MySQL 对比
11. 今日小结

1. MySQL 简介与安装

# MySQL 是最流行的开源关系型数据库
# 安装方式：

# Ubuntu/Debian
# sudo apt update && sudo apt install mysql-server

# CentOS/RHEL
# sudo yum install mysql-server

# macOS
# brew install mysql

# Docker（推荐学习用）
# docker run -d --name mysql -e MYSQL_ROOT_PASSWORD=123456 -p 3306:3306 mysql:8

# 安装 Python 驱动
# pip install pymysql

2. pymysql 安装与连接

# pip install pymysql
import pymysql

# 连接 MySQL
conn = pymysql.connect(
    host="localhost",       # 主机地址
    port=3306,              # 端口（默认3306）
    user="root",            # 用户名
    password="123456",      # 密码
    database="school",      # 数据库名
    charset="utf8mb4",      # 字符集（支持中文和 emoji）
    cursorclass=pymysql.cursors.DictCursor  # 返回字典格式
)

cursor = conn.cursor()

# 测试连接
cursor.execute("SELECT VERSION()")
print(f"MySQL 版本: {cursor.fetchone()['VERSION()']}")

conn.close()

💡 cursorclass=pymysql.cursors.DictCursor 让查询结果返回字典格式（如 {'name': '张三', 'age': 20}），比元组格式更易用。

3. 创建数据库和表

import pymysql

conn = pymysql.connect(
    host="localhost",
    user="root",
    password="123456",
    charset="utf8mb4"
)

cursor = conn.cursor()

# 创建数据库
cursor.execute("CREATE DATABASE IF NOT EXISTS school CHARACTER SET utf8mb4 COLLATE utf8mb4_unicode_ci")
cursor.execute("USE school")

# 创建表
cursor.execute("""
CREATE TABLE IF NOT EXISTS students (
    id INT AUTO_INCREMENT PRIMARY KEY,
    name VARCHAR(100) NOT NULL,
    age INT CHECK(age > 0),
    gender ENUM('男', '女', '未知') DEFAULT '未知',
    class VARCHAR(50) NOT NULL,
    email VARCHAR(200) UNIQUE,
    created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP
) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4
""")

conn.commit()
conn.close()
print("✅ 数据库和表创建成功")

4. 增删改查 (CRUD)

import pymysql

conn = pymysql.connect(
    host="localhost", user="root", password="123456",
    database="school", charset="utf8mb4",
    cursorclass=pymysql.cursors.DictCursor
)
cursor = conn.cursor()

# INSERT — 插入
cursor.execute(
    "INSERT INTO students (name, age, gender, class) VALUES (%s, %s, %s, %s)",
    ("张三", 20, "男", "一班")
)
print(f"插入 ID: {cursor.lastrowid}")

# 批量插入
students = [
    ("李四", 22, "女", "二班"),
    ("王五", 21, "男", "一班"),
    ("赵六", 23, "男", "二班"),
]
cursor.executemany(
    "INSERT INTO students (name, age, gender, class) VALUES (%s, %s, %s, %s)",
    students
)

conn.commit()

# SELECT — 查询
cursor.execute("SELECT * FROM students")
for row in cursor.fetchall():
    print(f"  {row['name']} | {row['age']}岁 | {row['class']}")

# UPDATE — 更新
cursor.execute("UPDATE students SET age = %s WHERE name = %s", (21, "张三"))
print(f"更新了 {cursor.rowcount} 行")

# DELETE — 删除
cursor.execute("DELETE FROM students WHERE name = %s", ("赵六",))
print(f"删除了 {cursor.rowcount} 行")

conn.commit()
conn.close()

⚠️ MySQL 用 %s 占位符，SQLite 用 ? 占位符！

5. 参数化查询 — 防 SQL 注入

import pymysql

conn = pymysql.connect(host="localhost", user="root", password="123456",
                       database="school", charset="utf8mb4",
                       cursorclass=pymysql.cursors.DictCursor)
cursor = conn.cursor()

# ❌ 危险！字符串格式化
# cursor.execute(f"SELECT * FROM students WHERE name = '{user_input}'")

# ✅ 安全！参数化查询
user_input = "张三' OR '1'='1"
cursor.execute("SELECT * FROM students WHERE name = %s", (user_input,))
print(f"安全查询: {cursor.fetchall()}")  # []（空结果）

# ✅ 也支持字典参数
cursor.execute(
    "SELECT * FROM students WHERE name = %(name)s AND age > %(min_age)s",
    {"name": "张三", "min_age": 18}
)

conn.close()

6. 事务管理

import pymysql

conn = pymysql.connect(host="localhost", user="root", password="123456",
                       database="school", charset="utf8mb4",
                       cursorclass=pymysql.cursors.DictCursor)
cursor = conn.cursor()

try:
    # 开始事务（pymysql 默认自动开启）
    cursor.execute("UPDATE accounts SET balance = balance - 100 WHERE name = '张三'")
    cursor.execute("UPDATE accounts SET balance = balance + 100 WHERE name = '李四'")

    # 提交事务
    conn.commit()
    print("✅ 转账成功")
except Exception as e:
    # 回滚事务
    conn.rollback()
    print(f"❌ 转账失败: {e}")
finally:
    conn.close()

7. 连接池

# DBUtils 提供连接池功能
# pip install dbutils

from dbutils.pooled_db import PooledDB
import pymysql

# 创建连接池
pool = PooledDB(
    creator=pymysql,        # 数据库驱动
    maxconnections=10,      # 最大连接数
    mincached=2,            # 最小空闲连接
    maxcached=5,            # 最大空闲连接
    host="localhost",
    user="root",
    password="123456",
    database="school",
    charset="utf8mb4",
    cursorclass=pymysql.cursors.DictCursor
)

# 从连接池获取连接
conn = pool.connection()
cursor = conn.cursor()
cursor.execute("SELECT * FROM students")
print(cursor.fetchall())
conn.close()  # 归还到连接池（不是真正关闭）

# 连接池的好处：
# 1. 避免频繁创建/销毁连接（开销大）
# 2. 控制最大连接数（防止数据库过载）
# 3. 自动管理连接生命周期

8. 错误处理

import pymysql

def safe_connect():
    """安全的数据库连接"""
    try:
        conn = pymysql.connect(
            host="localhost",
            user="root",
            password="wrong_password",
            database="school",
            charset="utf8mb4"
        )
        return conn
    except pymysql.err.OperationalError as e:
        print(f"❌ 连接失败: {e}")
        return None

def safe_execute(conn, sql, params=None):
    """安全执行 SQL"""
    try:
        cursor = conn.cursor()
        cursor.execute(sql, params)
        conn.commit()
        return cursor
    except pymysql.err.ProgrammingError as e:
        print(f"❌ SQL 语法错误: {e}")
        conn.rollback()
        return None
    except pymysql.err.IntegrityError as e:
        print(f"❌ 数据完整性错误（如主键重复）: {e}")
        conn.rollback()
        return None
    except pymysql.err.OperationalError as e:
        print(f"❌ 操作错误: {e}")
        conn.rollback()
        return None

9. 实战：用户管理系统

import pymysql
from datetime import datetime

class UserDB:
    """用户管理系统"""

    def __init__(self, host="localhost", user="root", password="123456", database="school"):
        self.conn = pymysql.connect(
            host=host, user=user, password=password,
            database=database, charset="utf8mb4",
            cursorclass=pymysql.cursors.DictCursor
        )
        self._create_table()

    def _create_table(self):
        self.conn.cursor().execute("""
        CREATE TABLE IF NOT EXISTS users (
            id INT AUTO_INCREMENT PRIMARY KEY,
            username VARCHAR(50) NOT NULL UNIQUE,
            email VARCHAR(200) NOT NULL,
            age INT,
            status TINYINT DEFAULT 1,
            created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP
        ) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4
        """)
        self.conn.commit()

    def create(self, username, email, age=None):
        """创建用户"""
        try:
            cursor = self.conn.cursor()
            cursor.execute(
                "INSERT INTO users (username, email, age) VALUES (%s, %s, %s)",
                (username, email, age)
            )
            self.conn.commit()
            return cursor.lastrowid
        except pymysql.err.IntegrityError:
            print(f"❌ 用户名 '{username}' 已存在")
            return None

    def get(self, user_id):
        """获取单个用户"""
        cursor = self.conn.cursor()
        cursor.execute("SELECT * FROM users WHERE id = %s", (user_id,))
        return cursor.fetchone()

    def list_all(self, page=1, page_size=10):
        """分页列出用户"""
        cursor = self.conn.cursor()
        offset = (page - 1) * page_size
        cursor.execute("SELECT * FROM users ORDER BY id LIMIT %s OFFSET %s", (page_size, offset))
        return cursor.fetchall()

    def update(self, user_id, **kwargs):
        """更新用户"""
        allowed = {"username", "email", "age", "status"}
        updates = {k: v for k, v in kwargs.items() if k in allowed}
        if not updates:
            return 0

        set_clause = ", ".join(f"{k} = %s" for k in updates)
        values = list(updates.values()) + [user_id]
        cursor = self.conn.cursor()
        cursor.execute(f"UPDATE users SET {set_clause} WHERE id = %s", values)
        self.conn.commit()
        return cursor.rowcount

    def delete(self, user_id):
        """删除用户"""
        cursor = self.conn.cursor()
        cursor.execute("DELETE FROM users WHERE id = %s", (user_id,))
        self.conn.commit()
        return cursor.rowcount

    def search(self, keyword):
        """搜索用户"""
        cursor = self.conn.cursor()
        cursor.execute(
            "SELECT * FROM users WHERE username LIKE %s OR email LIKE %s",
            (f"%{keyword}%", f"%{keyword}%")
        )
        return cursor.fetchall()

    def stats(self):
        """统计信息"""
        cursor = self.conn.cursor()
        cursor.execute("""
            SELECT 
                COUNT(*) as total,
                AVG(age) as avg_age,
                SUM(CASE WHEN status = 1 THEN 1 ELSE 0 END) as active
            FROM users
        """)
        return cursor.fetchone()

    def close(self):
        self.conn.close()

# 使用
db = UserDB()
db.create("alice", "alice@example.com", 25)
db.create("bob", "bob@example.com", 30)
db.create("charlie", "charlie@example.com", 22)

print("📋 所有用户:")
for u in db.list_all():
    print(f"  [{u['id']}] {u['username']} | {u['email']} | {u['age']}岁")

print(f"\n📊 统计: {db.stats()}")
db.close()

10. SQLite vs MySQL 对比

11. 今日小结

• ✅ pymysql.connect() 连接 MySQL
• ✅ MySQL 用 %s 占位符（SQLite 用 ?）
• ✅ cursorclass=DictCursor 返回字典格式
• ✅ 必须用参数化查询防 SQL 注入
• ✅ conn.commit() 提交，conn.rollback() 回滚
• ✅ 生产环境用连接池（DBUtils）
• ✅ 完善的错误处理

🎯 练习建议：
1. 给用户管理系统添加"批量导入"功能（从 CSV 导入）
2. 实现一个"数据库备份"工具（导出为 SQL 文件）
3. 对比 pymysql 和 sqlite3 的代码差异

📚 Day47 完成！明天学习 ORM — SQLAlchemy 入门