一、安装驱动

pip install mysql-connector-python

二、配置数据库

# 数据库连接信息（改成你自己的）

DB_CONFIG = {
    "host": "localhost",       # 本地一般是 localhost
    "user": "root",            # 你的 MySQL 用户名
    "password": "123456",      # 你的 MySQL 密码
    "database": "test_db",     # 要连接的数据库名
    "charset": "utf8mb4"       # 支持中文、表情
}

三、测试连接MySQL

import mysql.connector
from mysql.connector import Error

# 连接数据库
def connect_db():
    try:
        conn = mysql.connector.connect(**DB_CONFIG)
        if conn.is_connected():
            print("✅ 连接 MySQL 成功！")
            return conn
    except Error as e:
        print(f"❌ 连接失败：{e}")
        return None

# 测试
conn = connect_db()
if conn:
    conn.close()

四、查询所有/单条数据库

def query_all(sql, params=None):
    conn = connect_db()
    if not conn:
        return []
    cursor = conn.cursor(dictionary=True)  # 返回字典格式
    try:
        cursor.execute(sql, params)
        return cursor.fetchall()  # 获取所有数据
    except Error as e:
        print(f"查询失败：{e}")
        return []
    finally:
        cursor.close()
        conn.close()

# 使用示例
data = query_all("SELECT * FROM user")
for row in data:
    print(row)  # 每一行是字典，可直接 row["name"], row["age"]

五、新增数据库

def insert(sql, params):
    conn = connect_db()
    if not conn:
        return False
    cursor = conn.cursor()
    try:
        cursor.execute(sql, params)
        conn.commit()  # 必须提交！
        print(f"✅ 新增成功，影响行数：{cursor.rowcount}")
        return True
    except Error as e:
        conn.rollback()  # 失败回滚
        print(f"❌ 新增失败：{e}")
        return False
    finally:
        cursor.close()
        conn.close()

# 使用示例
insert(
    "INSERT INTO user(name, age) VALUES(%s, %s)",
    ("张三", 20)
)

六、修改数据

def update(sql, params):
    conn = connect_db()
    if not conn:
        return False
    cursor = conn.cursor()
    try:
        cursor.execute(sql, params)
        conn.commit()
        print(f"✅ 修改成功，影响行数：{cursor.rowcount}")
        return True
    except Error as e:
        conn.rollback()
        print(f"❌ 修改失败：{e}")
        return False
    finally:
        cursor.close()
        conn.close()

# 使用示例
update(
    "UPDATE user SET age=%s WHERE name=%s",
    (21, "张三")
)

七、删除数据

def delete(sql, params):
    conn = connect_db()
    if not conn:
        return False
    cursor = conn.cursor()
    try:
        cursor.execute(sql, params)
        conn.commit()
        print(f"✅ 删除成功，影响行数：{cursor.rowcount}")
        return True
    except Error as e:
        conn.rollback()
        print(f"❌ 删除失败：{e}")
        return False
    finally:
        cursor.close()
        conn.close()

# 使用示例
delete(
    "DELETE FROM user WHERE name=%s",
    ("张三",)
)