Python操作MySQL的从入门到实战指南

# 单行注释：用 # 开头，注释一行内容
# 下面是多行注释，用三个单引号或双引号包裹
'''
多行注释第一行
多行注释第二行
'''
"""
双引号版多行注释
适合注释较长的说明
"""

# 整数类型：用于存储年龄、编号等整数
age = 25
# 浮点数类型：用于存储小数
pi = 3.14
print(type(pi))  # 输出 <class 'float'>，查看数据类型

# 字符串类型：用于存储姓名、文本等
name = "maria"
print(type(name))  # 输出  <class 'str'>

# 输出方式示例
print("你好，Python！")  # 直接输出字符串
print(name)  # 输出变量
print("名字是：", name)  # 字符串和变量组合输出

score = 91

score = 91
if score >= 90:
    print("优秀")  # 满足条件时执行
elif score >= 80:
    print("良好")
elif score >= 60:
    print("及格")
else:
    print("不及格")  # 所有条件都不满足时执行

# for 循环：遍历字符串、列表等
name = "maria"
for char in name:
    print(char)  # 依次输出 m、a、r、i、a

# while 循环：根据条件重复执行
name = "maria"
index = 0
while index<len(name):  # len(name) 是字符串长度（5）
    char = name[index]
    print(char)
    index += 1  # 索引自增，避免死循环

# 定义函数：求两个数的最大值
def max_value(a, b):
    if a > b:
        return a  # 返回较大值
    else:
        return b

# 调用函数
result = max_value(2, 3)
print(result)  # 输出 3

-- 创建测试表 test_table
CREATE TABLE test_table (
  id INT AUTO_INCREMENT PRIMARY KEY,  # 自增主键
  name VARCHAR(255),  # 姓名
  age INT  # 年龄
);

-- 创建 Python 专用操作用户
CREATE USER 'python_rw'@'%' IDENTIFIED BY 'Idyq8_a1';
-- 授予增删查改等权限（根据实际需求调整）
GRANT create,alter,insert,delete,select,update ON *.* TO 'python_rw'@'%';

pip install pymysql==1.1.0

# 导入 pymysql 库
import pymysql
from pymysql.err import OperationalError, ProgrammingError

# 1. 建立数据库连接（替换为你的数据库信息）
# 增加连接参数，提升稳定性
config = {
    'host': '192.168.184.151',  # 数据库地址
    'user': 'python_rw',  # 用户名
    'password': 'Idyq8_a1',  # 密码
    'database': 'maria',  # 数据库名
    'charset': 'utf8mb4',  # 字符集（避免中文乱码）
    'connect_timeout': 10,  # 连接超时时间（防止卡死）
    'cursorclass': pymysql.cursors.DictCursor  # 可选：返回字典格式结果，更易读
}

# 初始化连接和游标为None，避免空对象调用
cnx = None
cursor = None

try:
    # 先测试连接是否能成功建立
    cnx = pymysql.connect(**config)
    print("✅ 数据库连接成功")

    # 2. 创建游标对象
    cursor = cnx.cursor()

    # 3. 插入数据（INSERT）
    sql_insert = "INSERT INTO test_table (name, age) VALUES (%s, %s)"
    values_insert = ("John", 25)
    cursor.execute(sql_insert, values_insert)
    cnx.commit()
    print(f"📥 数据插入成功，新增记录 ID：{cursor.lastrowid}")

    # 4. 查询数据（SELECT）
    sql_select = "SELECT * FROM test_table"
    cursor.execute(sql_select)
    rows = cursor.fetchall()
    print("\n📋 查询结果：")
    # 适配 DictCursor，用键取值（更直观）
    for row in rows:
        print(f"ID: {row['id']}, 姓名: {row['name']}, 年龄: {row['age']}")

    # 5. 更新数据（UPDATE）
    sql_update = "UPDATE test_table SET age = %s WHERE name = %s"
    values_update = (30, "John")
    cursor.execute(sql_update, values_update)
    cnx.commit()
    print(f"\n🔄 数据更新成功，影响行数：{cursor.rowcount}")

    # 6. 删除数据（DELETE）
    sql_delete = "DELETE FROM test_table WHERE name = %s"
    values_delete = ("John",)  # 单个参数保留逗号
    cursor.execute(sql_delete, values_delete)
    cnx.commit()
    print(f"🗑️  数据删除成功，影响行数：{cursor.rowcount}")

except OperationalError as e:
    # 专门捕获连接相关错误（地址/密码/数据库不存在）
    print(f"❌ 数据库连接失败：{str(e)}")
    print("请检查：1.数据库地址是否正确 2.用户名/密码是否正确 3.数据库是否存在")
except ProgrammingError as e:
    # 专门捕获SQL语法/表不存在错误
    print(f"❌ SQL执行失败：{str(e)}")
    print("请检查：1.test_table表是否存在 2.字段名（name/age）是否正确")
except Exception as e:
    # 其他未知错误
    if cnx:  # 只有连接成功才回滚
        cnx.rollback()
    print(f"❌ 操作失败：{str(e)}")
finally:
    # 7. 安全关闭游标和连接（判断对象是否存在，避免空指针）
    if cursor:
        cursor.close()
    if cnx:
        cnx.close()
    print("\n🔌 数据库连接已关闭")