Python 数据库访问

Python 数据库访问 (使用 MySQL Connector)

Python 通过 Python 数据库 API 规范 v2.0 (DB-API 2) 提供了一种与各种数据库交互的标准化方式。大多数 Python 数据库驱动/连接器都遵循这个标准。

你需要为你想要连接的数据库安装一个特定的符合 DB-API 标准的驱动（例如：PostgreSQL、SQLite、Oracle、SQL Server、MySQL）。本教程重点介绍使用官方的 mysql-connector-python 驱动访问 MySQL。

DB-API 的一般工作流程包括：

安装 mysql-connector-python

开始之前，使用 pip 安装驱动：

pip install mysql-connector-python

在 Python 解释器中尝试导入来验证安装：

import mysql.connector

如果运行没有出现 ImportError，则驱动已安装。

前提条件

对于下面的示例，假设：

建立数据库连接

使用 mysql.connector.connect() 建立连接。它返回一个连接对象。

import mysql.connector
from mysql.connector import Error

def create_connection(host_name, user_name, user_password, db_name):
    """Create a database connection to the MySQL database specified."""
    connection = None
    try:
        connection = mysql.connector.connect(
            host=host_name,
            user=user_name,
            password=user_password,
            database=db_name
        )
        print("MySQL Database connection successful")
    except Error as e:
        print(f"The error '{e}' occurred")
    return connection

# Example usage (replace with your details)
conn = create_connection("localhost", "testuser", "test1234", "testdb")

# Remember to close the connection when done
# if conn and conn.is_connected():
#    conn.close()
#    print("MySQL connection is closed")

推荐使用上下文管理器（with 语句）来处理连接和游标，因为它们确保资源会自动关闭。

创建数据库表

连接后，使用 connection.cursor() 创建一个游标对象。使用游标的 execute() 方法运行 SQL DDL (数据定义语言) 命令，如 CREATE TABLE。

import mysql.connector
from mysql.connector import Error

# Assume create_connection function exists as defined above

def execute_query(connection, query):
    """Executes a single SQL query."""
    cursor = connection.cursor()
    try:
        cursor.execute(query)
        connection.commit() # Important for DDL/DML if autocommit=False
        print("Query executed successfully")
    except Error as e:
        print(f"The error '{e}' occurred")
    finally:
        if cursor:
             cursor.close()

# -- Main execution --
conn = create_connection("localhost", "testuser", "test1234", "testdb")

if conn and conn.is_connected():
    # Drop table if it exists (for demonstration)
    drop_table_query = "DROP TABLE IF EXISTS employees;"
    execute_query(conn, drop_table_query)

    # Create the employees table
    create_employees_table = """
    CREATE TABLE employees (
      id INT AUTO_INCREMENT PRIMARY KEY,
      first_name VARCHAR(50) NOT NULL,
      last_name VARCHAR(50),
      age INT,
      gender CHAR(1),
      salary FLOAT
    );
    """
    execute_query(conn, create_employees_table)

    conn.close()
    print("MySQL connection is closed")

INSERT 操作 (创建记录)

使用带有 INSERT 语句的 execute() 方法。至关重要的是，使用参数化查询（占位符如 %s）来防止 SQL 注入漏洞。 将值作为元组或列表作为 execute() 的第二个参数传入。

# (Continuing from previous example, assuming connection 'conn' is open)

def insert_employee(connection, employee_data):
    """Inserts a new employee into the employees table."""
    sql = """
    INSERT INTO employees (first_name, last_name, age, gender, salary)
    VALUES (%s, %s, %s, %s, %s);
    """
    cursor = connection.cursor()
    try:
        cursor.execute(sql, employee_data)
        connection.commit() # Commit the transaction
        print(f"Employee {employee_data[0]} added successfully. ID: {cursor.lastrowid}")
        return cursor.lastrowid
    except Error as e:
        print(f"Failed to insert record: {e}")
        connection.rollback() # Rollback on error
    finally:
        if cursor:
            cursor.close()
    return None

# -- Inside main execution block --
if conn and conn.is_connected():
    # ... (table creation code) ...

    # Insert a single employee
    employee1 = ('Maria', 'Garcia', 30, 'F', 55000.00)
    insert_employee(conn, employee1)

    # Insert multiple employees using executemany
    employees_to_add = [
        ('John', 'Doe', 45, 'M', 75000.00),
        ('Lisa', 'Ray', 28, 'F', 62000.00)
    ]
    sql_insert_many = "INSERT INTO employees (first_name, last_name, age, gender, salary) VALUES (%s, %s, %s, %s, %s);"
    cursor = conn.cursor()
    try:
        cursor.executemany(sql_insert_many, employees_to_add)
        conn.commit()
        print(f"{cursor.rowcount} employees inserted successfully.")
    except Error as e:
        print(f"Failed to insert multiple records: {e}")
        conn.rollback()
    finally:
        if cursor:
            cursor.close()

    # ... (connection close) ...

READ 操作 (获取数据)

使用带有 SELECT 语句的 execute()。然后使用游标上的 fetch 方法，如 fetchone()、fetchall() 或 fetchmany(size) 来检索结果。

# (Continuing from previous example, assuming connection 'conn' is open)

def fetch_all_employees(connection):
    """Fetches all employee records."""
    sql = "SELECT id, first_name, last_name, age, salary FROM employees;"
    cursor = connection.cursor(dictionary=True) # Fetch as dictionaries
    employees = []
    try:
        cursor.execute(sql)
        employees = cursor.fetchall()
        print(f"Fetched {len(employees)} employee records.")
    except Error as e:
        print(f"Failed to fetch records: {e}")
    finally:
        if cursor:
            cursor.close()
    return employees

# -- Inside main execution block --
if conn and conn.is_connected():
    # ... (table creation, insertion code) ...

    all_employees = fetch_all_employees(conn)
    if all_employees:
        for employee in all_employees:
             # Access by column name if dictionary=True, else by index (e.g., employee[1])
            print(f"ID: {employee['id']}, Name: {employee['first_name']} {employee['last_name']}, Age: {employee['age']}")

    # ... (connection close) ...

在 connection.cursor(dictionary=True) 中设置 dictionary=True 会使 fetch 方法将行作为字典（列名 -> 值）返回，而不是元组，这通常更方便。

UPDATE 操作

使用带有 UPDATE 语句的 execute()。记住使用参数化查询并提交事务。

# (Continuing, assuming 'conn' is open)

def update_employee_salary(connection, employee_id, new_salary):
    """Updates the salary for a given employee ID."""
    sql = "UPDATE employees SET salary = %s WHERE id = %s;"
    cursor = connection.cursor()
    try:
        cursor.execute(sql, (new_salary, employee_id))
        connection.commit()
        if cursor.rowcount > 0:
             print(f"Salary updated for employee ID {employee_id}.")
        else:
             print(f"No employee found with ID {employee_id}.")
    except Error as e:
        print(f"Failed to update record: {e}")
        connection.rollback()
    finally:
        if cursor:
            cursor.close()

# -- Inside main execution block --
if conn and conn.is_connected():
    # ... (creation, insertion, fetch code) ...
    update_employee_salary(conn, 1, 60000.00) # Update salary for employee with ID 1
    # ... (connection close) ...

DELETE 操作

使用带有 DELETE 语句的 execute()。使用参数化查询并提交。

# (Continuing, assuming 'conn' is open)

def delete_employee(connection, employee_id):
    """Deletes an employee with the given ID."""
    sql = "DELETE FROM employees WHERE id = %s;"
    cursor = connection.cursor()
    try:
        cursor.execute(sql, (employee_id,))
        connection.commit()
        if cursor.rowcount > 0:
             print(f"Employee with ID {employee_id} deleted.")
        else:
             print(f"No employee found with ID {employee_id}.")
    except Error as e:
        print(f"Failed to delete record: {e}")
        connection.rollback()
    finally:
        if cursor:
            cursor.close()

# -- Inside main execution block --
if conn and conn.is_connected():
    # ... (creation, insertion, fetch, update code) ...
    delete_employee(conn, 3) # Delete employee with ID 3
    # ... (connection close) ...

事务 (commit() 和 rollback())

数据库事务使用 ACID 属性（原子性 Atomicity, 一致性 Consistency, 隔离性 Isolation, 持久性 Durability）来确保数据完整性。像 INSERT、UPDATE、DELETE 这样的操作通常是事务的一部分。

在成功的 DML 操作后提交更改或在发生错误时回滚至关重要，如 INSERT、UPDATE 和 DELETE 示例所示。默认情况下，mysql-connector-python 可能关闭了自动提交 (autocommit)，需要显式提交。

断开数据库连接 (close())

使用完毕后务必关闭数据库连接以释放资源。

if conn and conn.is_connected():
    conn.close()
    print("MySQL connection is closed")

使用 with 语句处理连接和游标会自动处理关闭，即使发生错误。

# Recommended pattern using 'with'
try:
    with create_connection("localhost", "testuser", "test1234", "testdb") as conn:
        if conn and conn.is_connected():
            with conn.cursor(dictionary=True) as cursor:
                cursor.execute("SELECT * FROM employees WHERE age > %s", (30,))
                results = cursor.fetchall()
                for row in results:
                    print(row)
            # Cursor is automatically closed here
        # Connection is automatically closed here (or rolled back on error within 'with')
        # Note: commit might still be needed explicitly depending on autocommit settings
except Error as e:
    print(f"Database error: {e}")
except Exception as e:
    print(f"An error occurred: {e}")

错误处理

数据库操作可能因各种原因失败（无效的 SQL、连接问题、约束违规）。使用 try...except 块来捕获错误。mysql-connector 模块会引发继承自 mysql.connector.Error 的异常。

根据需要捕获特定错误或基础的 Error 类。