目录

1. 引言：SQL 脚本执行的重要性
   • 1.1 为什么需要执行 SQL 脚本？
   • 1.2 Python sqlite3 模块概述
2. 核心方法：cursor.executescript()
   • 2.1 executescript() 的工作原理
   • 2.2 优势与特点
3. 分步实现：执行 SQL 脚本
   • 3.1 准备：创建示例 SQL 脚本文件
     • 脚本内容示例 (DDL, DML)
   • 3.2 连接到 SQLite 数据库
   • 3.3 读取 SQL 脚本内容
   • 3.4 使用 executescript() 执行脚本
   • 3.5 事务管理与提交
4. Python 代码示例
   • 4.1 创建示例 SQL 脚本
   • 4.2 执行脚本并验证
5. 最佳实践与注意事项
   • 5.1 错误处理 (try-except)
   • 5.2 脚本来源的安全性
   • 5.3 executescript() 的事务行为
   • 5.4 处理大型 SQL 脚本
   • 5.5 自定义脚本解析 (如果 executescript() 不满足需求)
6. 综合代码示例：完整的脚本执行与验证流程
7. 总结

---

1. 引言：SQL 脚本执行的重要性

在数据库管理和应用程序开发中，我们经常需要执行一系列相关的 SQL 语句。这些语句可能包括创建表、插入初始数据、修改表结构或定义存储过程等。将这些语句组织在一个单独的 SQL 脚本文件中，可以带来诸多好处：

1.1 为什么需要执行 SQL 脚本？

• 初始化数据库：在应用程序首次启动或部署时，用于创建所有必要的表、索引和视图，并填充初始数据。
• 数据库升级/迁移：管理数据库 schema 的版本，通过执行脚本来应用增量式的更改。
• 数据填充/导入：批量插入数据。
• 测试与重现：快速设置测试环境，或重现特定的数据库状态。
• 可维护性：将 SQL 逻辑从 Python 代码中分离出来，提高代码的可读性和维护性。

1.2 Python sqlite3 模块概述

Python 的 sqlite3 模块是标准库的一部分，提供了与 SQLite 数据库交互的接口。它遵循 Python DB-API 2.0 规范，提供了一致的数据库操作方法。

2. 核心方法：cursor.executescript()

sqlite3 模块为执行多条 SQL 语句的脚本提供了一个非常便捷的方法：cursor.executescript(sql_script_string)。

2.1 executescript() 的工作原理

此方法接受一个包含一个或多个 SQL 语句的字符串。它会自动将字符串根据语句分隔符（通常是分号 ;）拆分成单独的 SQL 语句，并逐一执行它们。

2.2 优势与特点

• 简化操作：无需手动解析 SQL 脚本，executescript() 自动处理语句的分割和执行。
• 内置事务：默认情况下，executescript() 会将其所有操作包装在一个隐式事务中。如果任何一个语句执行失败，整个事务通常会回滚（在 SQLite 3.6.8+ 中）。这确保了数据库状态的一致性。
• 易于使用：只需要读取文件内容并将其作为字符串传递给方法。

3. 分步实现：执行 SQL 脚本

3.1 准备：创建示例 SQL 脚本文件

我们首先需要一个 .sql 文件来演示。这个脚本可以包含 DDL (数据定义语言) 和 DML (数据操作语言) 语句。

示例 init_database.sql 文件内容：

-- init_database.sql
-- Drop tables if they exist to allow clean re-execution
DROP TABLE IF EXISTS employees;
DROP TABLE IF EXISTS departments;

-- Create departments table
CREATE TABLE departments (
    dept_id INTEGER PRIMARY KEY AUTOINCREMENT,
    dept_name TEXT NOT NULL UNIQUE
);

-- Create employees table
CREATE TABLE employees (
    emp_id INTEGER PRIMARY KEY AUTOINCREMENT,
    first_name TEXT NOT NULL,
    last_name TEXT NOT NULL,
    email TEXT UNIQUE,
    phone_number TEXT,
    hire_date DATE,
    job_id TEXT,
    salary REAL,
    dept_id INTEGER,
    FOREIGN KEY (dept_id) REFERENCES departments(dept_id)
);

-- Insert initial data into departments
INSERT INTO departments (dept_name) VALUES ('Sales');
INSERT INTO departments (dept_name) VALUES ('Marketing');
INSERT INTO departments (dept_name) VALUES ('Engineering');
INSERT INTO departments (dept_name) VALUES ('Human Resources');

-- Insert initial data into employees
INSERT INTO employees (first_name, last_name, email, hire_date, job_id, salary, dept_id) VALUES
('Alice', 'Smith', 'alice.s@example.com', '2022-01-15', 'SALES_REP', 60000.00, (SELECT dept_id FROM departments WHERE dept_name = 'Sales')),
('Bob', 'Johnson', 'bob.j@example.com', '2021-11-01', 'ENG_LEAD', 95000.00, (SELECT dept_id FROM departments WHERE dept_name = 'Engineering')),
('Charlie', 'Brown', 'charlie.b@example.com', '2023-03-20', 'MARKETING_COORD', 50000.00, (SELECT dept_id FROM departments WHERE dept_name = 'Marketing'));

-- Update an employee's salary
UPDATE employees SET salary = 62000.00 WHERE first_name = 'Alice';

3.2 连接到 SQLite 数据库

使用 sqlite3.connect() 方法建立数据库连接。通常，推荐使用 with 语句作为上下文管理器，以确保连接正确关闭。

import sqlite3
import os

DB_FILE = "company.db"

# with sqlite3.connect(DB_FILE) as conn:
#     # ... operations ...

3.3 读取 SQL 脚本内容

需要打开 .sql 文件并读取其所有内容到一个字符串变量中。

def read_sql_script(script_path):
    with open(script_path, 'r', encoding='utf-8') as f:
        sql_script = f.read()
    return sql_script

# sql_content = read_sql_script("init_database.sql")

3.4 使用 executescript() 执行脚本

一旦连接建立且脚本内容读取完毕，就可以通过游标对象调用 executescript()。

# cursor = conn.cursor()
# cursor.executescript(sql_content)

3.5 事务管理与提交

如前所述，executescript() 通常会将其操作包装在内部事务中。然而，最佳实践是在执行任何修改数据库的 DML/DDL 语句后，显式地调用 conn.commit() 来持久化这些更改。如果发生错误，可以通过 conn.rollback() 撤销未提交的更改。在使用 with sqlite3.connect(...) 时，如果没有未捕获的异常，上下文管理器会在退出时自动提交；如果有异常，则会自动回滚。

4. Python 代码示例

4.1 创建示例 SQL 脚本

import sqlite3
import os
import shutil # For cleanup

# --- Configuration ---
SQL_SCRIPT_FILE = "init_database.sql"
DB_FILE = "company.db"

def create_sample_sql_script(filename=SQL_SCRIPT_FILE):
    """Creates a sample SQL script file for demonstration."""
    print(f"Creating sample SQL script: {filename}")
    sql_content = """
    -- init_database.sql
    -- Drop tables if they exist to allow clean re-execution
    DROP TABLE IF EXISTS employees;
    DROP TABLE IF EXISTS departments;

    -- Create departments table
    CREATE TABLE departments (
        dept_id INTEGER PRIMARY KEY AUTOINCREMENT,
        dept_name TEXT NOT NULL UNIQUE
    );

    -- Create employees table
    CREATE TABLE employees (
        emp_id INTEGER PRIMARY KEY AUTOINCREMENT,
        first_name TEXT NOT NULL,
        last_name TEXT NOT NULL,
        email TEXT UNIQUE,
        phone_number TEXT,
        hire_date DATE,
        job_id TEXT,
        salary REAL,
        dept_id INTEGER,
        FOREIGN KEY (dept_id) REFERENCES departments(dept_id)
    );

    -- Insert initial data into departments
    INSERT INTO departments (dept_name) VALUES ('Sales');
    INSERT INTO departments (dept_name) VALUES ('Marketing');
    INSERT INTO departments (dept_name) VALUES ('Engineering');
    INSERT INTO departments (dept_name) VALUES ('Human Resources');

    -- Insert initial data into employees
    INSERT INTO employees (first_name, last_name, email, hire_date, job_id, salary, dept_id) VALUES
    ('Alice', 'Smith', 'alice.s@example.com', '2022-01-15', 'SALES_REP', 60000.00, (SELECT dept_id FROM departments WHERE dept_name = 'Sales')),
    ('Bob', 'Johnson', 'bob.j@example.com', '2021-11-01', 'ENG_LEAD', 95000.00, (SELECT dept_id FROM departments WHERE dept_name = 'Engineering')),
    ('Charlie', 'Brown', 'charlie.b@example.com', '2023-03-20', 'MARKETING_COORD', 50000.00, (SELECT dept_id FROM departments WHERE dept_name = 'Marketing'));

    -- Update an employee's salary
    UPDATE employees SET salary = 62000.00 WHERE first_name = 'Alice';
    """
    with open(filename, 'w', encoding='utf-8') as f:
        f.write(sql_content.strip())
    print(f"  Script '{filename}' created.")

def cleanup(script_filename=SQL_SCRIPT_FILE, db_filename=DB_FILE):
    """Cleans up generated script and database files."""
    print("\n--- Cleaning up ---")
    if os.path.exists(script_filename):
        os.remove(script_filename)
        print(f"  Deleted SQL script file: {script_filename}")
    if os.path.exists(db_filename):
        os.remove(db_filename)
        print(f"  Deleted database file: {db_filename}")

def read_sql_script(script_path):
    """Reads SQL script content from a file."""
    try:
        with open(script_path, 'r', encoding='utf-8') as f:
            return f.read()
    except FileNotFoundError:
        print(f"Error: SQL script file '{script_path}' not found.")
        return None
    except Exception as e:
        print(f"Error reading SQL script file: {e}")
        return None

4.2 执行脚本并验证

def execute_sql_script(db_filepath, script_filepath):
    """
    Connects to the database, reads and executes the SQL script.
    """
    print(f"\n--- Executing SQL script '{script_filepath}' on database '{db_filepath}' ---")
    sql_script_content = read_sql_script(script_filepath)
    if not sql_script_content:
        return False

    try:
        with sqlite3.connect(db_filepath) as conn:
            cursor = conn.cursor()
            cursor.executescript(sql_script_content)
            conn.commit() # Commit changes made by the script
            print("  SQL script executed successfully and changes committed.")
        return True
    except sqlite3.Error as e:
        print(f"  Database error during script execution: {e}")
        return False
    except Exception as e:
        print(f"  An unexpected error occurred: {e}")
        return False

def verify_database_state(db_filepath):
    """
    Verifies that tables and data were correctly created/inserted by the script.
    """
    print(f"\n--- Verifying database state for '{db_filepath}' ---")
    try:
        with sqlite3.connect(db_filepath) as conn:
            cursor = conn.cursor()

            # Verify departments table
            cursor.execute("SELECT COUNT(*) FROM departments;")
            dept_count = cursor.fetchone()[0]
            print(f"  Number of departments: {dept_count} (Expected: 4)")

            # Verify employees table
            cursor.execute("SELECT COUNT(*) FROM employees;")
            emp_count = cursor.fetchone()[0]
            print(f"  Number of employees: {emp_count} (Expected: 3)")

            # Verify specific data (e.g., Alice's updated salary)
            cursor.execute("SELECT salary FROM employees WHERE first_name = 'Alice';")
            alice_salary = cursor.fetchone()[0]
            print(f"  Alice's salary: {alice_salary} (Expected: 62000.0)")

            # Fetch all employees to display
            cursor.execute("SELECT emp_id, first_name, last_name, salary, dept_name FROM employees JOIN departments ON employees.dept_id = departments.dept_id ORDER BY emp_id;")
            employees_data = cursor.fetchall()
            print("\n  Current Employees:")
            print(f"{'ID':<5} {'First Name':<15} {'Last Name':<15} {'Salary':<10} {'Department':<15}")
            print(f"{'-'*5:<5} {'-'*15:<15} {'-'*15:<15} {'-'*10:<10} {'-'*15:<15}")
            for emp in employees_data:
                print(f"{emp[0]:<5} {emp[1]:<15} {emp[2]:<15} {emp[3]:<10.2f} {emp[4]:<15}")

    except sqlite3.Error as e:
        print(f"  Database error during verification: {e}")
    except Exception as e:
        print(f"  An unexpected error occurred during verification: {e}")

# --- Main execution flow ---
# cleanup() # Ensure clean start
# create_sample_sql_script()
# if execute_sql_script(DB_FILE, SQL_SCRIPT_FILE):
#     verify_database_state(DB_FILE)
# cleanup() # Clean up after execution

5. 最佳实践与注意事项

5.1 错误处理 (try-except)

在文件读取和数据库操作时，始终使用 try...except 块来捕获潜在的 FileNotFoundError (读取 SQL 脚本时) 和 sqlite3.Error (数据库操作失败时)。这使得程序更健壮。

5.2 脚本来源的安全性

如果 SQL 脚本的路径或内容来自不可信的外部输入（如用户上传），那么执行该脚本可能存在严重的安全风险（SQL 注入、数据泄露、系统破坏等）。务必对脚本内容进行严格的验证和沙箱隔离，或者只执行受信任的、预定义的脚本。在本示例中，我们假设脚本是安全的。

5.3 executescript() 的事务行为

如前所述，executescript() 默认会将整个脚本作为一个事务来执行。这意味着如果脚本中的任何一条语句失败，所有先前的更改都可能被回滚。这对于保持数据库一致性非常有用。然而，如果你的脚本非常庞大且某些部分是独立的，你可能需要手动分割脚本并分批执行和提交，以避免单个大型事务。

5.4 处理大型 SQL 脚本

对于包含数千甚至数百万条语句的巨型 SQL 脚本：

• 内存消耗：一次性将整个脚本读入内存可能会消耗大量内存。
• 性能：单个巨型事务可能会导致长时间的锁定。
• 替代方案：可以考虑自己解析脚本，逐条或分批执行。或者，如果是在 Unix-like 系统上，可以直接通过 subprocess 调用 sqlite3 命令行工具来执行脚本，这通常更适合超大文件。

5.5 自定义脚本解析 (如果 executescript() 不满足需求)

在极少数情况下，如果 executescript() 的默认行为（如语句分隔符处理或事务管理）不符合你的特定需求，你可以自己实现一个脚本解析器。这通常涉及：

1. 读取脚本内容。
2. 根据分号 (;) 或其他自定义分隔符将内容分割成单个 SQL 语句。
3. 遍历语句列表，使用 cursor.execute() 逐条执行。
4. 手动管理事务 (conn.commit() / conn.rollback())。
   但对于大多数情况，executescript() 已经足够。

6. 综合代码示例：完整的脚本执行与验证流程

import sqlite3
import os
import shutil # For cleanup

# --- Configuration ---
SQL_SCRIPT_FILE = "init_database_full.sql"
DB_FILE = "company_full.db"

# --- 1. Helper Functions ---
def create_sample_sql_script(filename=SQL_SCRIPT_FILE):
    """Creates a sample SQL script file for demonstration."""
    print(f"\n--- Creating sample SQL script: {filename} ---")
    sql_content = """
    -- init_database_full.sql
    -- Drop tables if they exist to allow clean re-execution
    DROP TABLE IF EXISTS employees;
    DROP TABLE IF EXISTS departments;

    -- Create departments table
    CREATE TABLE departments (
        dept_id INTEGER PRIMARY KEY AUTOINCREMENT,
        dept_name TEXT NOT NULL UNIQUE
    );

    -- Create employees table
    CREATE TABLE employees (
        emp_id INTEGER PRIMARY KEY AUTOINCREMENT,
        first_name TEXT NOT NULL,
        last_name TEXT NOT NULL,
        email TEXT UNIQUE,
        phone_number TEXT,
        hire_date DATE,
        job_id TEXT,
        salary REAL,
        dept_id INTEGER,
        FOREIGN KEY (dept_id) REFERENCES departments(dept_id)
    );

    -- Insert initial data into departments
    INSERT INTO departments (dept_name) VALUES ('Sales');
    INSERT INTO departments (dept_name) VALUES ('Marketing');
    INSERT INTO departments (dept_name) VALUES ('Engineering');
    INSERT INTO departments (dept_name) VALUES ('Human Resources');

    -- Insert initial data into employees
    INSERT INTO employees (first_name, last_name, email, hire_date, job_id, salary, dept_id) VALUES
    ('Alice', 'Smith', 'alice.s@example.com', '2022-01-15', 'SALES_REP', 60000.00, (SELECT dept_id FROM departments WHERE dept_name = 'Sales')),
    ('Bob', 'Johnson', 'bob.j@example.com', '2021-11-01', 'ENG_LEAD', 95000.00, (SELECT dept_id FROM departments WHERE dept_name = 'Engineering')),
    ('Charlie', 'Brown', 'charlie.b@example.com', '2023-03-20', 'MARKETING_COORD', 50000.00, (SELECT dept_id FROM departments WHERE dept_name = 'Marketing'));

    -- Update an employee's salary
    UPDATE employees SET salary = 62000.00 WHERE first_name = 'Alice';
    """
    with open(filename, 'w', encoding='utf-8') as f:
        f.write(sql_content.strip())
    print(f"  SQL script '{filename}' created.")

def cleanup_files(script_filename=SQL_SCRIPT_FILE, db_filename=DB_FILE):
    """Deletes generated script and database files."""
    print("\n--- Cleaning up files ---")
    if os.path.exists(script_filename):
        os.remove(script_filename)
        print(f"  Deleted SQL script file: {script_filename}")
    if os.path.exists(db_filename):
        os.remove(db_filename)
        print(f"  Deleted database file: {db_filename}")

def read_sql_script_content(script_path):
    """Reads SQL script content from a file."""
    try:
        with open(script_path, 'r', encoding='utf-8') as f:
            return f.read()
    except FileNotFoundError:
        print(f"Error: SQL script file '{script_path}' not found. Cannot proceed.")
        return None
    except Exception as e:
        print(f"Error reading SQL script file '{script_path}': {e}. Cannot proceed.")
        return None

def execute_sql_script(db_filepath, script_filepath):
    """
    Connects to the database, reads and executes the SQL script using executescript().
    Includes comprehensive error handling.
    """
    print(f"\n--- Executing SQL script '{script_filepath}' on database '{db_filepath}' ---")
    sql_script_content = read_sql_script_content(script_filepath)
    if not sql_script_content:
        return False # Script content could not be read

    conn = None # Initialize conn to None for error handling
    try:
        conn = sqlite3.connect(db_filepath)
        cursor = conn.cursor()
        
        # Executescript handles splitting multiple SQL statements and implicit transactions.
        # If an error occurs during script execution, the implicit transaction will typically rollback.
        cursor.executescript(sql_script_content)
        
        conn.commit() # Explicitly commit all changes made by the script
        print("  SQL script executed successfully and changes committed.")
        return True
    except sqlite3.Error as e:
        print(f"  Database error during script execution: {e}")
        if conn:
            conn.rollback() # Rollback in case of database error
            print("  Transaction rolled back due to error.")
        return False
    except Exception as e:
        print(f"  An unexpected error occurred: {e}")
        if conn:
            conn.rollback()
            print("  Transaction rolled back due to unexpected error.")
        return False
    finally:
        if conn:
            conn.close() # Ensure connection is closed even if an error occurs

def verify_database_state(db_filepath):
    """
    Verifies that tables and data were correctly created/inserted by the script.
    """
    print(f"\n--- Verifying database state for '{db_filepath}' ---")
    conn = None
    try:
        conn = sqlite3.connect(db_filepath)
        conn.row_factory = sqlite3.Row # Allows accessing columns by name
        cursor = conn.cursor()

        # Verify departments table
        cursor.execute("SELECT COUNT(*) FROM departments;")
        dept_count = cursor.fetchone()[0]
        print(f"  Number of departments: {dept_count} (Expected: 4)")

        # Verify employees table
        cursor.execute("SELECT COUNT(*) FROM employees;")
        emp_count = cursor.fetchone()[0]
        print(f"  Number of employees: {emp_count} (Expected: 3)")

        # Verify specific data (e.g., Alice's updated salary)
        cursor.execute("SELECT first_name, last_name, salary FROM employees WHERE first_name = 'Alice';")
        alice_data = cursor.fetchone()
        if alice_data:
            print(f"  Alice's salary: {alice_data['salary']:.2f} (Expected: 62000.0)")
        else:
            print("  Alice not found in employees table.")


        # Fetch all employees with department name to display
        cursor.execute("""
            SELECT emp_id, first_name, last_name, salary, dept_name 
            FROM employees 
            JOIN departments ON employees.dept_id = departments.dept_id 
            ORDER BY emp_id;
        """)
        employees_data = cursor.fetchall()
        print("\n  Current Employees in Database:")
        if employees_data:
            print(f"{'ID':<5} {'First Name':<15} {'Last Name':<15} {'Salary':<10} {'Department':<15}")
            print(f"{'-'*5:<5} {'-'*15:<15} {'-'*15:<15} {'-'*10:<10} {'-'*15:<15}")
            for emp in employees_data:
                print(f"{emp['emp_id']:<5} {emp['first_name']:<15} {emp['last_name']:<15} {emp['salary']:<10.2f} {emp['dept_name']:<15}")
        else:
            print("  No employees found in the database.")

    except sqlite3.Error as e:
        print(f"  Database error during verification: {e}")
    except Exception as e:
        print(f"  An unexpected error occurred during verification: {e}")
    finally:
        if conn:
            conn.close()

# --- Main Program Execution ---
def main():
    cleanup_files() # Ensure a clean slate before starting
    create_sample_sql_script() # Create the SQL script file

    if execute_sql_script(DB_FILE, SQL_SCRIPT_FILE):
        verify_database_state(DB_FILE)
    
    print("\n--- Program finished ---")
    # cleanup_files() # Uncomment this line to delete the generated files after execution

if __name__ == "__main__":
    main()

7. 总结

为您详尽解析了在 Python 中如何使用 sqlite3 模块来执行包含多条 SQL 语句的脚本文件。

核心要点回顾:

• cursor.executescript(sql_script_string) 是执行 SQL 脚本的核心方法，它会自动解析并执行脚本中的所有语句。
• 脚本准备：将一系列 SQL 语句编写到 .sql 文件中，并使用 Python 读取其内容。
• 事务管理：executescript() 默认将脚本包装在一个事务中，并通过 conn.commit() (或 with 语句的自动提交) 进行持久化。
• 错误处理：务必使用 try-except sqlite3.Error 来捕获数据库错误，并在必要时进行 conn.rollback()。
• 安全性：对于来自外部的 SQL 脚本，务必验证其内容以防止安全风险。

通过掌握 cursor.executescript() 方法，您将能够高效、安全地在 Python 应用程序中初始化、升级或批量操作 SQLite 数据库，极大地提高了数据库管理的便捷性。