MySQL - Python 语法

现代 Python 与 MySQL 集成

将 Python 应用程序连接到 MySQL 数据库是任何开发人员的基本技能。这通过使用“连接器（connector）”或“驱动程序（driver）”来实现，它们是实现了 Python 标准数据库 API 规范 v2.0 (DB-API 2.0) 的专用库。这个 API 确保了从 Python 与不同数据库交互时感觉一致且可预测。

设置开发环境

在编写代码之前，建立一个干净且隔离的开发环境至关重要。这可以防止项目依赖之间的冲突。

步骤 1：创建虚拟环境

使用虚拟环境是一个不可妥协的最佳实践。在项目文件夹中打开终端并运行：

# 适用于 macOS/Linux
python3 -m venv venv
source venv/bin/activate

# 适用于 Windows
python -m venv venv
.\venv\Scripts\activate

您会看到 (venv) 前缀出现在命令提示符前，表示虚拟环境已激活。

步骤 2：安装 MySQL 连接器

Oracle 官方连接器是 mysql-connector-python。使用 pip 安装它：

pip install mysql-connector-python

替代方案：许多开发者更喜欢 PyMySQL（pip install PyMySQL），因为它是一个纯 Python 实现，在某些环境中可能更容易安装。其 API 几乎相同，便于切换。

数据库交互生命周期

在 Python 中与数据库交互遵循一个清晰的模式：

1. 连接（Connect）：建立与 MySQL 服务器的连接。
2. 游标（Cursor）：从连接中创建一个游标对象。游标就像一个控制器，负责执行查询和获取结果。
3. 执行（Execute）：使用游标执行 SQL 查询，使用占位符来防止 SQL 注入。
4. 获取（Fetch）：如果执行的是 SELECT 语句，则从执行的查询中检索结果。
5. 提交/回滚（Commit/Rollback）：如果您进行了更改（INSERT、UPDATE、DELETE），则必须 commit() 它们以永久保存。如果发生错误，则 rollback() 以撤销当前事务中的更改。
6. 关闭（Close）：关闭游标和连接以释放数据库资源。

关键函数

函数	描述
connect()	建立与 MySQL 服务器的连接。接受主机、用户、密码等作为参数。
connection.cursor()	创建一个游标对象。您可以从一个连接创建多个游标。
cursor.execute(query, params)	执行 SQL 查询。params 参数（元组或列表）用于安全地替换参数。
cursor.fetchone()	从结果集中获取下一行作为元组。
cursor.fetchall()	从结果集中获取所有剩余行作为元组列表。
connection.commit()	保存自上次提交以来所做的所有数据更改。
connection.rollback()	放弃自上次提交以来所做的所有数据更改。
cursor.close() / connection.close()	关闭游标和连接。这样做对于释放服务器资源至关重要。

包含最佳实践的完整 CRUD 示例

此示例演示了一个完整的创建（Create）、读取（Read）、更新（Update）和删除（Delete）(CRUD) 周期，其中包含了安全连接处理和参数化查询等基本最佳实践。

安全警示：切勿在源代码中硬编码凭据（用户名、密码）。请使用环境变量或安全的配置系统。此示例使用 os.getenv() 进行演示。

import mysql.connector
from mysql.connector import errorcode
import os

# --- 配置 ---
def get_db_connection():
    """创建并返回一个数据库连接。"""
    try:
        connection = mysql.connector.connect(
            host=os.getenv('DB_HOST', 'localhost'),
            user=os.getenv('DB_USER', 'root'),
            password=os.getenv('DB_PASSWORD', 'password'),
            database='your_database' # 替换为您的数据库名称
        )
        return connection
    except mysql.connector.Error as err:
        if err.errno == errorcode.ER_ACCESS_DENIED_ERROR:
            print("Something is wrong with your user name or password")
        elif err.errno == errorcode.ER_BAD_DB_ERROR:
            print("Database does not exist")
        else:
            print(err)
        return None

# --- 主应用程序逻辑 ---
def main():
    """运行主要的 CRUD 操作。"""
    connection = get_db_connection()
    if not connection:
        return

    cursor = connection.cursor()

    try:
        # 1. 创建（CREATE）：设置一个表
        print("\n--- Creating 'products' table ---")
        cursor.execute("DROP TABLE IF EXISTS products")
        cursor.execute("""
            CREATE TABLE products (
                id INT AUTO_INCREMENT PRIMARY KEY,
                name VARCHAR(255) NOT NULL,
                price DECIMAL(10, 2) NOT NULL
            )
        """)
        print("Table 'products' created successfully.")

        # 2. 插入（INSERT）：使用参数化查询添加新数据
        print("\n--- Inserting new products ---")
        new_products = [
            ('Laptop', 1200.50),
            ('Mouse', 25.00),
            ('Keyboard', 75.99)
        ]
        insert_query = "INSERT INTO products (name, price) VALUES (%s, %s)"
        cursor.executemany(insert_query, new_products)
        connection.commit() # 提交插入操作
        print(f"{cursor.rowcount} products inserted.")

        # 3. 读取（READ）：选择并显示数据
        print("\n--- Reading all products ---")
        cursor.execute("SELECT id, name, price FROM products")
        for (id, name, price) in cursor:
            print(f"ID: {id}, Name: {name}, Price: ${price:.2f}")

        # 4. 更新（UPDATE）：修改现有记录
        print("\n--- Updating product price ---")
        update_query = "UPDATE products SET price = %s WHERE name = %s"
        cursor.execute(update_query, (29.99, 'Mouse'))
        connection.commit() # 提交更新操作
        print("Mouse price updated.")

        # 5. 删除（DELETE）：移除记录
        print("\n--- Deleting a product ---")
        delete_query = "DELETE FROM products WHERE name = %s"
        cursor.execute(delete_query, ('Keyboard',))
        connection.commit() # 提交删除操作
        print("Keyboard deleted.")

        # 验证最终状态
        print("\n--- Final product list ---")
        cursor.execute("SELECT id, name, price FROM products")
        rows = cursor.fetchall()
        for row in rows:
            print(row)

    except mysql.connector.Error as err:
        print(f"An error occurred: {err}")
        connection.rollback() # 错误时回滚更改
    finally:
        # 6. 关闭（CLOSE）：务必关闭游标和连接
        print("\n--- Closing resources ---")
        cursor.close()
        connection.close()
        print("Connection closed.")

if __name__ == '__main__':
    main()