Python如何操作MySQL数据库？连接与查询实战

mysql-connector-python

pip install PyMySQL

import pymysql

# 数据库连接配置，这些信息需要根据你的实际情况来填
DB_CONFIG = {
    'host': '127.0.0.1', # 或者你的MySQL服务器IP
    'port': 3306,
    'user': 'your_username',
    'password': 'your_password',
    'db': 'your_database_name',
    'charset': 'utf8mb4', # 确保支持表情符号等
    'cursorclass': pymysql.cursors.DictCursor # 这样查询结果会是字典，方便取值
}

conn = None
cursor = None
try:
    # 建立连接
    conn = pymysql.connect(**DB_CONFIG)
    # 获取游标对象，用于执行SQL命令
    cursor = conn.cursor()

    # --- 1. 查询数据 ---
    sql_select = "SELECT id, name, age FROM users WHERE age > %s"
    # 使用参数化查询，防止SQL注入
    cursor.execute(sql_select, (25,))
    # 获取所有查询结果
    users = cursor.fetchall()
    print("查询结果:")
    for user in users:
        print(f"ID: {user['id']}, Name: {user['name']}, Age: {user['age']}")

    # --- 2. 插入数据 ---
    sql_insert = "INSERT INTO users (name, age) VALUES (%s, %s)"
    new_user_data = [
        ('Alice', 30),
        ('Bob', 22)
    ]
    # 使用executemany批量插入，效率更高
    cursor.executemany(sql_insert, new_user_data)
    # 提交事务，让更改生效
    conn.commit()
    print(f"\n成功插入 {cursor.rowcount} 条数据。")

    # --- 3. 更新数据 ---
    sql_update = "UPDATE users SET age = %s WHERE name = %s"
    cursor.execute(sql_update, (31, 'Alice'))
    conn.commit()
    print(f"\n成功更新 {cursor.rowcount} 条数据。")

    # --- 4. 删除数据 ---
    sql_delete = "DELETE FROM users WHERE name = %s"
    cursor.execute(sql_delete, ('Bob',))
    conn.commit()
    print(f"\n成功删除 {cursor.rowcount} 条数据。")

except pymysql.Error as e:
    print(f"数据库操作失败: {e}")
    if conn:
        conn.rollback() # 发生错误时回滚事务
except Exception as e:
    print(f"发生未知错误: {e}")
finally:
    # 确保连接和游标被关闭，释放资源
    if cursor:
        cursor.close()
    if conn:
        conn.close()
    print("\n数据库连接已关闭。")

host

port

user

password

utf8mb4

charset='utf8mb4'

conn.close()

cursor.close()

finally

sql = "SELECT id, name FROM users WHERE age > %s"

cursor.execute(sql, (25,))

INSERT INTO users (name, age) VALUES (%s, %s)

executemany()

execute()

fetchall()

with

import pymysql

# ... DB_CONFIG ...

try:
    with pymysql.connect(**DB_CONFIG) as conn:
        with conn.cursor() as cursor:
            # 在这里执行你的查询和操作
            cursor.execute("SELECT VERSION()")
            version = cursor.fetchone()
            print(f"MySQL Version: {version}")
            # conn.commit() 在with块结束时，如果没异常且conn.autocommit是False，会自动提交
except pymysql.Error as e:
    print(f"数据库操作失败: {e}")
    # with 语句会处理回滚，如果autocommit是False

commit()

rollback()

import pymysql

# ... DB_CONFIG ...

conn = None
try:
    conn = pymysql.connect(**DB_CONFIG)
    # 关闭自动提交，这样所有的操作都需要手动commit或rollback
    conn.autocommit(False)

    cursor = conn.cursor()

    # 假设一个转账场景：从用户1001扣100元，给用户1002加100元
    # 步骤1: 扣款
    sql_deduct = "UPDATE accounts SET balance = balance - %s WHERE user_id = %s AND balance >= %s"
    cursor.execute(sql_deduct, (100, 1001, 100))
    if cursor.rowcount == 0:
        raise ValueError("用户1001余额不足或用户不存在，无法扣款。")

    print("成功从用户1001扣款100元。")

    # 模拟一个可能出错的场景，比如网络中断或逻辑错误
    # if True: # 假设这里有个条件，满足时会出错
    #     raise Exception("模拟转账过程中发生意外错误！")

    # 步骤2: 加款
    sql_add = "UPDATE accounts SET balance = balance + %s WHERE user_id = %s"
    cursor.execute(sql_add, (100, 1002))
    if cursor.rowcount == 0:
        raise ValueError("用户1002不存在，无法加款。")

    print("成功给用户1002加款100元。")

    # 如果所有操作都成功，提交事务
    conn.commit()
    print("事务提交成功，转账完成。")

except pymysql.Error as e:
    print(f"数据库操作失败，回滚事务: {e}")
    if conn:
        conn.rollback() # 发生任何数据库错误时回滚
except ValueError as e:
    print(f"业务逻辑错误，回滚事务: {e}")
    if conn:
        conn.rollback() # 业务逻辑错误也回滚
except Exception as e:
    print(f"发生未知错误，回滚事务: {e}")
    if conn:
        conn.rollback() # 其他任何异常也回滚
finally:
    if cursor:
        cursor.close()
    if conn:
        conn.close()
    print("数据库连接已关闭。")

conn.autocommit(False)

conn.commit()

conn.rollback()

rollback