Python怎样操作MySQL数据库？PyMySQL连接方法

import pymysql

# 数据库连接参数
db_config = {
    'host': 'localhost',
    'user': 'your_username',
    'password': 'your_password',
    'database': 'your_database',
    'charset': 'utf8mb4', # 推荐使用utf8mb4以支持更广泛的字符集
    'cursorclass': pymysql.cursors.DictCursor # 推荐使用DictCursor，结果以字典形式返回
}

conn = None
try:
    # 建立连接
    conn = pymysql.connect(**db_config)
    print("数据库连接成功！")

    # 创建游标对象，用于执行SQL查询
    # 使用with语句可以确保游标在代码块结束时自动关闭
    with conn.cursor() as cursor:
        # 示例：创建表（如果不存在）
        create_table_sql = """
        CREATE TABLE IF NOT EXISTS users (
            id INT AUTO_INCREMENT PRIMARY KEY,
            name VARCHAR(255) NOT NULL,
            email VARCHAR(255) UNIQUE NOT NULL
        ) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4;
        """
        cursor.execute(create_table_sql)
        print("表 'users' 检查或创建完成。")

        # 示例：插入数据
        # 注意：使用参数化查询来防止SQL注入
        insert_sql = "INSERT INTO users (name, email) VALUES (%s, %s)"
        users_to_insert = [
            ("张三", "zhangsan@example.com"),
            ("李四", "lisi@example.com")
        ]
        for user_data in users_to_insert:
            try:
                cursor.execute(insert_sql, user_data)
                print(f"插入数据: {user_data[0]} 成功。")
            except pymysql.err.IntegrityError as e:
                # 捕获唯一键冲突等错误
                print(f"插入数据: {user_data[0]} 失败，可能已存在或数据冲突: {e}")
            except Exception as e:
                print(f"插入数据: {user_data[0]} 发生未知错误: {e}")

        # 提交事务，使更改生效
        conn.commit()
        print("数据插入事务已提交。")

        # 示例：查询数据
        select_sql = "SELECT id, name, email FROM users WHERE name = %s"
        cursor.execute(select_sql, ("张三",))
        result = cursor.fetchone() # 获取一条结果
        if result:
            print(f"查询到用户张三: ID={result['id']}, Name={result['name']}, Email={result['email']}")

        # 示例：查询所有数据
        cursor.execute("SELECT id, name, email FROM users")
        all_users = cursor.fetchall() # 获取所有结果
        print("\n所有用户：")
        for user in all_users:
            print(f"ID={user['id']}, Name={user['name']}, Email={user['email']}")

except pymysql.err.OperationalError as e:
    # 捕获连接或操作数据库时的错误
    print(f"数据库操作错误: {e}")
    if conn:
        conn.rollback() # 发生错误时回滚事务
        print("事务已回滚。")
except Exception as e:
    # 捕获其他未知错误
    print(f"发生未知错误: {e}")
    if conn:
        conn.rollback()
        print("事务已回滚。")
finally:
    # 确保连接最终被关闭
    if conn:
        conn.close()
        print("数据库连接已关闭。")

from DBUtils.PooledDB import PooledDB
import pymysql

# 数据库配置，与之前类似
db_config = {
    'host': 'localhost',
    'user': 'your_username',
    'password': 'your_password',
    'database': 'your_database',
    'charset': 'utf8mb4',
    'cursorclass': pymysql.cursors.DictCursor
}

# 初始化连接池，通常在应用启动时执行一次
# mincached: 启动时创建的空闲连接数
# maxcached: 连接池中允许的最大空闲连接数
# maxconnections: 连接池允许的最大连接数（包括使用中和空闲的）
# blocking: 如果连接池已满，是否阻塞等待
pool = PooledDB(
    creator=pymysql, # 指定底层数据库模块
    mincached=5,
    maxcached=10,
    maxconnections=20,
    blocking=True,
    ping=1, # 检查连接是否可用，如果断开会自动重新连接
    **db_config
)

def execute_query_with_pool(sql, params=None):
    conn = None
    cursor = None
    result = None
    try:
        # 从连接池获取连接
        conn = pool.connection()
        with conn.cursor() as cursor:
            cursor.execute(sql, params)
            if sql.strip().upper().startswith("SELECT"):
                result = cursor.fetchall()
            else:
                conn.commit()
                result = cursor.rowcount # 返回影响的行数
    except Exception as e:
        print(f"数据库操作错误: {e}")
        if conn:
            conn.rollback() # 发生错误时回滚
        raise # 重新抛出异常，让调用者处理
    finally:
        if conn:
            # 连接被归还到连接池，而不是真正关闭
            conn.close()
    return result

# 示例使用
try:
    # 插入数据
    execute_query_with_pool("INSERT INTO users (name, email) VALUES (%s, %s)", ("王五", "wangwu@example.com"))
    print("王五插入成功。")

    # 查询数据
    users = execute_query_with_pool("SELECT * FROM users")
    print("\n通过连接池查询到的所有用户:")
    for user in users:
        print(user)
except Exception as e:
    print(f"操作失败: {e}")

# 千万不要这样做！
user_input_name = "张三'; DROP TABLE users;--" # 恶意输入
sql = f"SELECT * FROM users WHERE name = '{user_input_name}'"
cursor.execute(sql)

# 推荐做法
user_input_name = "张三" # 或者任何用户输入
sql = "SELECT * FROM users WHERE name = %s" # 使用%s作为占位符
cursor.execute(sql, (user_input_name,)) # 将参数作为元组传递给execute方法

conn = None
try:
    conn = pymysql.connect(**db_config)
    # 默认autocommit=False，所以需要手动提交
    with conn.cursor() as cursor:
        # 操作1：更新用户余额
        cursor.execute("UPDATE accounts SET balance = balance - %s WHERE user_id = %s", (100, 1))
        # 模拟一个可能失败的操作
        # 如果这里发生错误，下面的commit就不会执行
        # 操作2：记录交易日志
        cursor.execute("INSERT INTO transactions (user_id, amount, type) VALUES (%s, %s, %s)", (1, -100, 'expense'))

    conn.commit() # 所有操作都成功，提交事务
    print("事务提交成功，数据已更新。")

except Exception as e:
    print(f"事务失败: {e}")
    if conn:
        conn.rollback() # 任何一个操作失败，回滚所有更改
        print("事务已回滚，数据未更改。")
finally:
    if conn:
        conn.close()