如何通过Python 3.6代码调用SQLite命令行批量执行CSV导入命令以提升数据导入效率？

Absolutely! You can absolutely automate SQLite's fast CSV import workflow using Python by calling the SQLite command-line tool directly. This lets you leverage SQLite's built-in .import command—way faster than row-by-row inserts via the sqlite3 module, especially for large datasets. Here's a step-by-step solution:

Core Approach: Use subprocess to Call SQLite CLI #

Python's subprocess module lets you execute system commands, including the SQLite command-line tool. We'll use it to send the .mode csv and .import commands directly, avoiding slow Python-level data handling.

Full Code Example #

This script will scan a folder for all CSV files, import each into a corresponding table in your database.db, and include optimizations to speed up imports even more:

import subprocess
import os
import glob
import csv

DB_PATH = "database.db"
CSV_FOLDER = "./your_csv_directory"  # Replace with your CSV folder path

# Iterate over all CSV files in the target folder
for csv_path in glob.glob(os.path.join(CSV_FOLDER, "*.csv")):
    # Generate a safe table name from the CSV filename (remove .csv extension)
    csv_filename = os.path.basename(csv_path)
    table_name = os.path.splitext(csv_filename)[0]
    # Handle special characters in table names (like spaces or hyphens)
    table_name = f"`{table_name}`"

    # Optional: If your CSV has a header row, read it to define table columns
    # Skip this block if you want SQLite to auto-create the table with TEXT columns
    with open(csv_path, "r", newline="", encoding="utf-8") as f:
        reader = csv.reader(f)
        headers = next(reader)
        # Define columns as TEXT (adjust types if needed for your data)
        columns = ", ".join([f"`{h}` TEXT" for h in headers])
        create_table_sql = f"CREATE TABLE IF NOT EXISTS {table_name} ({columns});"

    # Build the sequence of SQLite commands
    # Optimizations: Disable sync/foreign keys temporarily for speed
    sqlite_commands = f"""
PRAGMA synchronous=OFF;
PRAGMA foreign_keys=OFF;
PRAGMA journal_mode=MEMORY;
{create_table_sql}  # Omit this line if skipping the header handling block
.mode csv
.import '{csv_path}' --skip 1 {table_name}  # Use --skip 1 to ignore header row
PRAGMA synchronous=NORMAL;
PRAGMA foreign_keys=ON;
"""

    # Execute the commands via SQLite CLI
    process = subprocess.Popen(
        ["sqlite3", DB_PATH],
        stdin=subprocess.PIPE,
        stdout=subprocess.PIPE,
        stderr=subprocess.PIPE,
        text=True
    )
    stdout, stderr = process.communicate(input=sqlite_commands)

    # Check for errors
    if process.returncode != 0:
        print(f"Failed to import {csv_filename}: {stderr.strip()}")
    else:
        print(f"Successfully imported {csv_filename} into table {table_name}")

Key Details & Notes #

• Table Name Safety: We wrap table names in backticks to handle special characters (like spaces, hyphens, or reserved words).
• Header Handling: The --skip 1 flag (available in SQLite 3.32.0+) tells .import to ignore the first row (your CSV header). If your SQLite version is older, you'll need to manually skip the header row or create the table first then import the rest of the data.
• Speed Optimizations: The PRAGMA commands temporarily disable disk syncing, foreign key checks, and use an in-memory journal—this drastically reduces disk I/O during imports. We re-enable these settings afterward for database safety.
• Security: We avoid using shell=True to prevent shell injection risks. Instead, we pass commands via stdin which is safer.

Alternative: Optimize sqlite3 Module Inserts (If You Don't Want to Call CLI) #

If you prefer to stick with Python's sqlite3 module instead of calling the CLI, you can still speed up inserts significantly by:

• Using executemany() instead of individual execute() calls
• Wrapping inserts in a transaction
• Disabling the same PRAGMA optimizations mentioned above

However, this will still be slower than SQLite's native .import command for very large files, since it's still processing data through Python.

内容的提问来源于stack exchange，提问作者bunirules