以下将针对新手用户,用「详细文字说明+分步代码」的方式演示全流程,重点满足「插入10条数据」「操作前后打印已有信息」「通俗解释每步意义」的需求,确保零基础也能看懂。

准备工作说明

在开始前,先简单解释几个核心概念(新手必看):

  • MySQL:一种常用的数据库软件,用于存储和管理数据(类似Excel,但功能更强大、适合大量数据)。
  • PyMySQL:Python连接MySQL的工具,让我们能用Python代码操作数据库。
  • 数据库(DB):相当于一个「文件夹」,里面可以放多个「表」。
  • 表(Table):相当于「Excel表格」,有行(记录)和列(字段),比如我们要创建的testTable会有id(编号)、name(姓名)、age(年龄)三列。
  • 游标(Cursor):执行SQL命令的「工具」,相当于我们操作鼠标点击Excel的动作。

步骤1:连接MySQL,查看当前已有数据库,然后创建testDB

目的:先看看MySQL里本来有哪些数据库,再创建我们需要的testDB,最后验证是否创建成功。

import pymysql

# 连接MySQL的基本信息(请确认你的MySQL主机、账号、密码正确)
config = {
    "host": "localhost",  # 数据库地址,本地一般是localhost
    "user": "admin",      # 你的用户名
    "password": "digital_history666",  # 你的密码
    "port": 3306,         # 端口,默认3306,不用改
    "charset": "utf8mb4"  # 支持中文,固定写法
}

# 定义一个函数:查询当前所有数据库(方便多次调用)
def show_databases(cursor):
    cursor.execute("SHOW DATABASES")  # 执行查询所有数据库的SQL
    dbs = cursor.fetchall()  # 获取查询结果
    print("当前已有的数据库列表:")
    for db in dbs:
        print(db[0])  # 打印数据库名(结果是元组,取第一个元素)

# 开始操作
conn = None  # 连接对象,先初始化为空
cursor = None  # 游标对象,先初始化为空
try:
    # 第一步:连接MySQL(此时还没进入任何数据库,因为要先看所有库)
    conn = pymysql.connect(** config)
    cursor = conn.cursor()  # 创建游标,用于执行SQL
    print("="*50)
    print("【操作前】查看当前MySQL中的所有数据库:")
    show_databases(cursor)  # 调用函数,打印已有数据库

    # 第二步:创建testDB数据库(IF NOT EXISTS表示如果已存在就不重复创建,避免报错)
    create_db_sql = "CREATE DATABASE IF NOT EXISTS testDB"
    cursor.execute(create_db_sql)  # 执行创建数据库的命令
    print("\n【创建数据库】执行了创建testDB的命令")

    # 第三步:再次查看数据库,确认testDB已创建
    print("\n【操作后】再次查看所有数据库,确认testDB是否存在:")
    show_databases(cursor)
    print("\n✅ 步骤1完成:testDB数据库已准备好(若已存在则忽略创建)")

except pymysql.Error as e:
    # 如果出错,打印错误原因(新手常见错误:密码错、MySQL没启动、账号没权限)
    print(f"\n❌ 步骤1出错:{e}")
    print("可能原因:MySQL服务未启动;账号密码错误;admin账号没有创建数据库的权限")
finally:
    # 无论成功失败,都要关闭游标和连接(释放资源,好习惯)
    if cursor:
        cursor.close()
    if conn:
        conn.close()
    print("="*50 + "\n")

运行后你会看到:

  • 操作前的数据库列表(比如默认的mysql、information_schema等)。
  • 操作后列表中会新增testdb(MySQL不区分大小写,显示可能是小写),说明创建成功。

步骤2:进入testDB,查看当前已有表,然后创建testTable

目的:进入testDB数据库后,先看看里面有没有表,再创建我们需要的testTable,最后验证表是否创建成功。

# 定义一个函数:查询当前数据库中的所有表(方便多次调用)
def show_tables(cursor):
    cursor.execute("SHOW TABLES")  # 执行查询所有表的SQL
    tables = cursor.fetchall()  # 获取结果
    print("当前数据库中的表列表:")
    for table in tables:
        print(table[0])  # 打印表名

# 开始操作
conn = None
cursor = None
try:
    # 第一步:连接并进入testDB数据库(这次要指定database参数)
    db_config = config.copy()  # 复制基础配置
    db_config["database"] = "testDB"  # 指定要操作的数据库
    conn = pymysql.connect(**db_config)
    cursor = conn.cursor()
    print("="*50)
    print("【操作前】查看testDB数据库中已有的表:")
    show_tables(cursor)  # 刚创建的数据库通常没有表,会显示空列表

    # 第二步:创建testTable表(定义表结构)
    # 表包含3列:
    # id:自动增长的编号(主键,唯一标识每条记录,类似Excel的行号)
    # name:姓名(非空,必须填)
    # age:年龄(可以为空)
    create_table_sql = """
    CREATE TABLE IF NOT EXISTS testTable (
        id INT PRIMARY KEY AUTO_INCREMENT,
        name VARCHAR(50) NOT NULL,
        age INT
    )
    """
    cursor.execute(create_table_sql)  # 执行创建表的命令
    print("\n【创建表】执行了创建testTable的命令,表结构:id(编号)、name(姓名)、age(年龄)")

    # 第三步:再次查看表,确认testTable已创建
    print("\n【操作后】再次查看testDB中的表:")
    show_tables(cursor)
    print("\n✅ 步骤2完成:testTable表已准备好(若已存在则忽略创建)")

except pymysql.Error as e:
    print(f"\n❌ 步骤2出错:{e}")
    print("可能原因:testDB数据库不存在(请先运行步骤1);没有创建表的权限")
finally:
    if cursor:
        cursor.close()
    if conn:
        conn.close()
    print("="*50 + "\n")

运行后你会看到:

  • 操作前testDB中没有表(列表为空)。
  • 操作后会显示testtable,说明表创建成功。

步骤3:向testTable插入10条数据,插入后查看结果

目的:往testTable中添加10条模拟数据(比如不同姓名和年龄),插入后查看是否成功。

# 开始操作
conn = None
cursor = None
try:
    # 连接到testDB
    db_config = config.copy()
    db_config["database"] = "testDB"
    conn = pymysql.connect(** config)
    cursor = conn.cursor()  # 普通游标(返回元组),后面查询用字典游标更直观

    # 第一步:先查看表中现有数据(刚创建的表应该为空)
    print("="*50)
    print("【插入前】testTable中的数据(应为空):")
    cursor.execute("SELECT * FROM testTable")
    before_data = cursor.fetchall()
    print(before_data)  # 空表会打印()

    # 第二步:准备10条数据(姓名用"用户1"到"用户10",年龄随机18-30岁)
    # 数据格式:(name, age),用列表存起来
    data_list = [
        ("用户1", 18), ("用户2", 19), ("用户3", 20), ("用户4", 21), ("用户5", 22),
        ("用户6", 23), ("用户7", 24), ("用户8", 25), ("用户9", 26), ("用户10", 27)
    ]

    # 第三步:插入数据(用executemany批量插入,比循环单条插入效率高)
    insert_sql = "INSERT INTO testTable (name, age) VALUES (%s, %s)"
    # %s是占位符,避免SQL注入(安全写法!不要直接用字符串拼接)
    cursor.executemany(insert_sql, data_list)  # 批量插入所有数据
    conn.commit()  # 提交事务!增删改必须执行这一步,否则数据不生效(关键!)

    # 第四步:插入后查看数据(用字典游标,结果带字段名,更易读)
    cursor = conn.cursor(pymysql.cursors.DictCursor)  # 切换为字典游标
    cursor.execute("SELECT * FROM testTable")
    after_data = cursor.fetchall()  # 获取所有数据
    print("\n【插入后】testTable中的10条数据(带字段名):")
    for row in after_data:
        print(row)  # 格式:{'id': 1, 'name': '用户1', 'age': 18}

    print(f"\n✅ 步骤3完成:成功插入{len(data_list)}条数据")

except pymysql.Error as e:
    if conn:
        conn.rollback()  # 出错时回滚,避免数据混乱
    print(f"\n❌ 步骤3出错:{e}")
    print("可能原因:表结构不对;数据格式错误(比如name为空)")
finally:
    if cursor:
        cursor.close()
    if conn:
        conn.close()
    print("="*50 + "\n")

关键说明:

  • executemany:批量插入多条数据,比循环调用execute更高效。
  • conn.commit():必须执行!MySQL默认需要手动确认修改,否则插入的数据不会保存(就像Excel的「保存」按钮)。
  • 字典游标:返回的结果是字典(带id/name/age字段名),比默认的元组更直观,适合新手。

步骤4:查询testTable中的数据(多种查询方式)

目的:演示如何查询数据(比如查所有数据、按条件查、查前几条),让新手了解查询的灵活性。

# 开始操作
conn = None
cursor = None
try:
    # 连接到testDB,直接用字典游标(方便查看字段名)
    db_config = config.copy()
    db_config["database"] = "testDB"
    conn = pymysql.connect(** db_config)
    cursor = conn.cursor(pymysql.cursors.DictCursor)
    print("="*50)
    print("【查询演示】")

    # 1. 查询所有数据(确认总条数)
    cursor.execute("SELECT * FROM testTable")
    all_data = cursor.fetchall()
    print(f"\n1. 所有数据(共{len(all_data)}条):")
    for row in all_data[:3]:  # 只打印前3条,避免太长
        print(row)
    if len(all_data) > 3:
        print("...(省略部分数据)")

    # 2. 按条件查询(比如查询年龄≥25的用户)
    cursor.execute("SELECT * FROM testTable WHERE age >= %s", (25,))  # %s传参
    age_data = cursor.fetchall()
    print(f"\n2. 年龄≥25的用户(共{len(age_data)}条):")
    for row in age_data:
        print(row)

    # 3. 查询指定字段(比如只查name和age,不查id)
    cursor.execute("SELECT name, age FROM testTable WHERE id <= 5")
    field_data = cursor.fetchall()
    print(f"\n3. id≤5的用户(只显示name和age):")
    for row in field_data:
        print(row)

    print("\n✅ 步骤4完成:查询演示结束")

except pymysql.Error as e:
    print(f"\n❌ 步骤4出错:{e}")
    print("可能原因:表名或字段名写错(比如把age写成ag)")
finally:
    if cursor:
        cursor.close()
    if conn:
        conn.close()
    print("="*50 + "\n")

新手注意:

  • 查询时用%s传参(比如age >= %s),然后在execute的第二个参数传入具体值(元组格式,如(25,)),这是安全的写法,防止SQL注入(一种黑客攻击方式)。
  • fetchall()会获取所有结果,如果数据量大,可用fetchmany(5)获取前5条,或fetchone()逐条获取。

步骤5:修改testTable中的数据(更新操作)

目的:演示如何修改已有数据(比如把「用户1」的年龄改成30),修改后查看结果。

# 开始操作
conn = None
cursor = None
try:
    db_config = config.copy()
    db_config["database"] = "testDB"
    conn = pymysql.connect(**config)
    cursor = conn.cursor(pymysql.cursors.DictCursor)
    print("="*50)

    # 第一步:修改前先查看要改的数据(比如用户1的原始年龄)
    cursor.execute("SELECT * FROM testTable WHERE name = %s", ("用户1",))
    before_update = cursor.fetchone()  # 获取一条数据
    print(f"【修改前】用户1的信息:{before_update}")

    # 第二步:执行修改(把用户1的年龄改成30,id改成100)
    update_sql = "UPDATE testTable SET age = %s, id = %s WHERE name = %s"
    # 参数:(新年龄, 新id, 姓名)
    cursor.execute(update_sql, (30, 100, "用户1"))
    conn.commit()  # 必须提交!否则修改不生效

    # 第三步:修改后查看结果
    cursor.execute("SELECT * FROM testTable WHERE name = %s", ("用户1",))
    after_update = cursor.fetchone()
    print(f"【修改后】用户1的信息:{after_update}")
    print(f"修改影响的行数:{cursor.rowcount}(正常应为1)")  # rowcount显示改了几行

    print("\n✅ 步骤5完成:数据修改成功")

except pymysql.Error as e:
    if conn:
        conn.rollback()  # 出错回滚
    print(f"\n❌ 步骤5出错:{e}")
    print("可能原因:id被重复修改(主键必须唯一);条件写错导致没找到数据")
finally:
    if cursor:
        cursor.close()
    if conn:
        conn.close()
    print("="*50 + "\n")

关键说明:

  • UPDATE语句必须加WHERE条件!否则会修改表中所有数据(比如UPDATE testTable SET age=30会把所有人的年龄改成30,新手容易犯这个错)。
  • rowcount:显示本次修改影响了多少行,可用来确认是否修改到了目标数据。

步骤6:删除testTable中的数据(删除操作)

目的:演示如何删除数据(比如删除「用户10」),删除后查看结果。

# 开始操作
conn = None
cursor = None
try:
    db_config = config.copy()
    db_config["database"] = "testDB"
    conn = pymysql.connect(** db_config)
    cursor = conn.cursor(pymysql.cursors.DictCursor)
    print("="*50)

    # 第一步:删除前先查看所有数据(确认用户10存在)
    cursor.execute("SELECT * FROM testTable")
    before_delete = cursor.fetchall()
    print(f"【删除前】总条数:{len(before_delete)}条(应包含用户10)")

    # 第二步:执行删除(删除姓名为"用户10"的数据)
    delete_sql = "DELETE FROM testTable WHERE name = %s"
    cursor.execute(delete_sql, ("用户10",))
    conn.commit()  # 必须提交!

    # 第三步:删除后查看结果
    cursor.execute("SELECT * FROM testTable")
    after_delete = cursor.fetchall()
    print(f"【删除后】总条数:{len(after_delete)}条(应减少1条)")
    print("删除后的数据(前3条):")
    for row in after_delete[:3]:
        print(row)

    print(f"\n删除影响的行数:{cursor.rowcount}(正常应为1)")
    print("\n✅ 步骤6完成:数据删除成功")

except pymysql.Error as e:
    if conn:
        conn.rollback()  # 出错回滚
    print(f"\n❌ 步骤6出错:{e}")
    print("可能原因:条件写错导致没找到要删除的数据")
finally:
    if cursor:
        cursor.close()
    if conn:
        conn.close()
    print("="*50 + "\n")

重要警告:

  • DELETE语句必须加WHERE条件!否则会删除表中所有数据(非常危险!新手一定要注意)。
  • 删除后的数据很难恢复,实际开发中建议先备份,或用「逻辑删除」(比如加一个is_deleted字段标记是否删除,而不是真的删除行)。

总结

通过以上6个步骤,我们完成了从「创建数据库→创建表→插入10条数据→查询→修改→删除」的全流程,每个步骤都包含「操作前检查→执行操作→操作后验证」,方便新手理解每一步的效果。

如果运行中遇到错误,先看错误提示(比如「Access denied」是权限问题,「Unknown database」是数据库没创建),再对照步骤排查配置或SQL语句是否正确。

更多推荐