PyMySQL 入门：用 Python 操作 MySQL 数据库

一、为什么要用 Python 操作数据库？

• 手动用 SQL 命令行操作 → 适合临时查询
• 用 Python 程序操作 → 适合批量处理、自动化测试、Web 系统后台
• 典型场景：
  • 自动化测试前，自动往数据库插入测试数据
  • 测试完成后，自动清理脏数据
  • 写一个小脚本，定期备份某张表

1.1 pymysql 是什么？

• pymysql 是一个纯 Python 写的 MySQL 客户端库
• 用法和 MySQL 命令行非常像：连接 → 执行 SQL → 拿结果 → 关闭
• 安装：

pip install pymysql

---

二、第一步：连接数据库

2.1 最基本的连接

import pymysql

# 1. 建立连接（相当于 mysql -u root -p 登录后进入某个数据库）
conn = pymysql.connect(
    host="localhost",      # 数据库地址
    user="root",           # 用户名
    password="123456",     # 密码
    database="school_db",  # 要操作的数据库
    charset="utf8mb4"      # 字符集，推荐 utf8mb4
)

# 2. 关闭连接（用完一定要关，不然会占用资源）
conn.close()

2.2 容易踩的坑（错误 vs 正确）

错误写法：密码、数据库名写错，连接时不处理异常，程序直接崩

# ❌ 不处理异常
conn = pymysql.connect(
    host="localhost",    
    user="root", 
    ’password="xxx", 
    database="school_db")
# 一旦密码错，程序直接报错退出

正确写法：用 try/except 包住连接过程

# ✅ 捕获连接异常
import pymysql

try:
    conn = pymysql.connect(
        host="localhost",
        user="root",
        password="123456",
        database="school_db",
        charset="utf8mb4"
    )
    print("数据库连接成功！")
except pymysql.err.OperationalError as e:
    print(f"数据库连接失败：{e}")
finally:
    # finally 里的代码不管成功失败都会执行，适合做清理
    if 'conn' in dir() and conn.open:
        conn.close()
        print("连接已关闭")

---

三、核心对象：Connection 和 Cursor

可以把这两个对象想象成现实场景：

对象	类比	作用
Connection	手机和基站的通话	负责连接数据库、管理事务
Cursor	你的嘴巴和耳朵	负责发送 SQL、接收查询结果

一个连接上可以创建多个游标，每个游标独立执行 SQL。

conn = pymysql.connect(...)
cursor = conn.cursor()   # 创建一个游标

---

四、查询数据：SELECT

4.1 查询所有学生

import pymysql

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

try:
    cursor = conn.cursor()

    # 编写 SQL
    sql = "SELECT id, name, age, gender FROM student"
    cursor.execute(sql)    # 执行 SQL

    # 获取结果
    rows = cursor.fetchall()   # 拿到所有行
    for row in rows:
        print(row)
finally:
    cursor.close()
    conn.close()

输出类似：

(1, '张三', 20, '男')
(2, '李四', 21, '男')
(3, '王五', 19, '女')
...

4.2 获取结果的几种方式

方法	作用
fetchone()	获取下一行，没有返回 None
fetchmany(n)	获取接下来 n 行
fetchall()	获取所有剩余行

4.3 让结果变成字典（推荐新手用）

默认结果是一个元组 (1, '张三', 20, '男')，字段含义要自己对应，不方便。加一个参数就能变成字典：

cursor = conn.cursor(cursor=pymysql.cursors.DictCursor)

cursor.execute("SELECT id, name, age FROM student")
for row in cursor.fetchall():
    print(f"学号：{row['id']}，姓名：{row['name']}，年龄：{row['age']}")

输出：

学号：1，姓名：张三，年龄：20
学号：2，姓名：李四，年龄：21
...

4.4 带条件的查询（参数化）

错误写法：用字符串拼接，容易产生 SQL 注入漏洞

# ❌ 危险！用户输入可能包含恶意 SQL
name = "张三' OR '1'='1"
sql = f"SELECT * FROM student WHERE name = '{name}'"

正确写法：使用 %s 占位符，让 pymysql 自动处理转义

# ✅ 安全
name = "张三"
sql = "SELECT * FROM student WHERE name = %s"
cursor.execute(sql, (name,))   # 注意：参数是一个元组，哪怕只有一个值也要加逗号

---

五、插入数据：INSERT

5.1 插入一条数据

import pymysql

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

try:
    cursor = conn.cursor()

    sql = """
        INSERT INTO student (name, age, gender, phone, class_id)
        VALUES (%s, %s, %s, %s, %s)
    """
    cursor.execute(sql, ("陈十一", 22, "男", "13800001234", 2))

    # ⚠️ 增删改操作必须 commit 才会真正保存到数据库
    conn.commit()
    print(f"插入成功，影响的行数：{cursor.rowcount}")
    print(f"新记录的 id：{cursor.lastrowid}")
except Exception as e:
    print(f"插入失败：{e}")
    conn.rollback()   # 出错就回滚，保证数据一致性
finally:
    cursor.close()
    conn.close()

5.2 批量插入多条

sql = """
    INSERT INTO student (name, age, gender, phone, class_id)
    VALUES (%s, %s, %s, %s, %s)
"""
students = [
    ("赵一", 20, "男", "13800000001", 1),
    ("钱二", 21, "女", "13800000002", 2),
    ("孙三", 19, "男", "13800000003", 3),
]

cursor.executemany(sql, students)   # 注意是 executemany
conn.commit()
print(f"批量插入 {cursor.rowcount} 条")

⚠️ 关键区别：execute() 执行一条，executemany() 执行多条。增删改都要 commit()，查询不需要。

---

六、更新数据：UPDATE

sql = "UPDATE student SET age = %s WHERE name = %s"
cursor.execute(sql, (25, "张三"))
conn.commit()
print(f"更新了 {cursor.rowcount} 条记录")

---

七、删除数据：DELETE

sql = "DELETE FROM student WHERE name = %s"
cursor.execute(sql, ("陈十一",))
conn.commit()
print(f"删除了 {cursor.rowcount} 条记录")

⚠️ 练习删除时，先用 SELECT 看一下会被删掉哪些行，避免误删。

---

八、完整示例：学生信息管理小程序

把上面学的知识组合起来，写一个小工具：

import pymysql


def get_conn():
    """获取数据库连接"""
    return pymysql.connect(
        host="localhost",
        user="root",
        password="123456",
        database="school_db",
        charset="utf8mb4",
        cursorclass=pymysql.cursors.DictCursor
    )


def list_students():
    """查询所有学生"""
    conn = get_conn()
    try:
        with conn.cursor() as cursor:
            cursor.execute("""
                SELECT s.id, s.name, s.age, s.gender, c.class_name
                FROM student s
                LEFT JOIN class c ON s.class_id = c.id
                ORDER BY s.id
            """)
            for row in cursor.fetchall():
                print(f"[{row['id']}] {row['name']} | {row['age']}岁 | "
                      f"{row['gender']} | {row['class_name']}")
    finally:
        conn.close()


def add_student(name, age, gender, phone, class_id):
    """新增一个学生"""
    conn = get_conn()
    try:
        with conn.cursor() as cursor:
            sql = """
                INSERT INTO student (name, age, gender, phone, class_id)
                VALUES (%s, %s, %s, %s, %s)
            """
            cursor.execute(sql, (name, age, gender, phone, class_id))
            conn.commit()
            print(f"新增成功，学号：{cursor.lastrowid}")
    except Exception as e:
        conn.rollback()
        print(f"新增失败：{e}")
    finally:
        conn.close()


def update_student_age(student_id, new_age):
    """修改学生年龄"""
    conn = get_conn()
    try:
        with conn.cursor() as cursor:
            sql = "UPDATE student SET age = %s WHERE id = %s"
            cursor.execute(sql, (new_age, student_id))
            conn.commit()
            print(f"修改完成，影响 {cursor.rowcount} 条")
    except Exception as e:
        conn.rollback()
        print(f"修改失败：{e}")
    finally:
        conn.close()


def delete_student(student_id):
    """删除学生"""
    conn = get_conn()
    try:
        with conn.cursor() as cursor:
            # 先查一下是否存在
            cursor.execute("SELECT name FROM student WHERE id = %s", (student_id,))
            row = cursor.fetchone()
            if not row:
                print(f"学号 {student_id} 不存在")
                return
            cursor.execute("DELETE FROM student WHERE id = %s", (student_id,))
            conn.commit()
            print(f"已删除：{row['name']}")
    except Exception as e:
        conn.rollback()
        print(f"删除失败：{e}")
    finally:
        conn.close()


# 测试一下
if __name__ == "__main__":
    print("=== 学生列表 ===")
    list_students()

    print("\n=== 新增学生 ===")
    add_student("小明", 20, "男", "13912345678", 2)

    print("\n=== 再次查询 ===")
    list_students()

---

九、事务：保证一组操作要么全成功，要么全失败

场景：转账——A 的账户扣 100 元，B 的账户加 100 元。两步必须一起成功，否则钱就"丢了"。

try:
    cursor.execute("UPDATE account SET balance = balance - 100 WHERE name = 'A'")
    cursor.execute("UPDATE account SET balance = balance + 100 WHERE name = 'B'")
    conn.commit()      # 两条 SQL 一起提交
    print("转账成功")
except Exception as e:
    conn.rollback()    # 任何一步出错，全部撤销
    print(f"转账失败：{e}")

要点：

• commit()：提交，让改动真正生效
• rollback()：回滚，撤销本次事务里所有操作
• 查询（SELECT）不需要 commit，增删改都需要

---

十、上下文管理器（with 语句）简化代码

每次都要写 try/finally/close 很麻烦，可以用 with 简化：

with pymysql.connect(host="localhost", user="root",
                     password="123456", database="school_db",
                     charset="utf8mb4") as conn:
    with conn.cursor(cursor=pymysql.cursors.DictCursor) as cursor:
        cursor.execute("SELECT * FROM student")
        for row in cursor.fetchall():
            print(row)
# 出了 with 会自动关闭 cursor 和 connection

小结

• pymysql 是 Python 操作 MySQL 的常用库
• 基本流程：连接 → 游标 → 执行 SQL → 拿结果 → 关闭
• 增删改必须 commit，查询不需要
• 参数化查询用 %s，永远不要字符串拼接用户输入
• 用 DictCursor 让结果变成字典，更易读
• 用 try/except/finally 或 with 保证资源正确关闭