使用 Python 将 CSV 文件导入 SQLite 数据库
·
目录
- 引言：CSV 与 SQLite 的桥梁
- 1.1 为什么需要导入 CSV 到 SQLite？
- 1.2 Python
sqlite3与csv模块简介
- 准备工作：理解 CSV 结构与 SQLite 数据类型
- 2.1 CSV 文件结构概述
- 2.2 SQLite 数据类型亲和性 (Type Affinity)
- 核心步骤解析
- 3.1 创建示例 CSV 文件
- 3.2 读取 CSV 文件：Python
csv模块csv.reader与csv.DictReader的选择
- 3.3 创建 SQLite 表结构
- 方法一：手动定义表结构 (预知 schema)
- 方法二：动态从 CSV 头部推断表结构 (更灵活)
- 3.4 将数据插入 SQLite 表
- 单行插入的局限性
- 批量插入 (
executemany) - 推荐 - 事务管理的重要性
- 分步代码实现
- 4.1 准备：创建示例 CSV 文件
- 4.2 方法一：使用
csv.reader和预定义表结构 - 4.3 方法二：使用
csv.DictReader和动态表结构 (推荐且更通用)
- 最佳实践与注意事项
- 5.1 错误处理 (文件、数据库、数据)
- 5.2 数据类型转换 (字符串到数值/日期)
- 5.3 性能优化：批量插入和提交频率
- 5.4 编码问题 (
encoding参数) - 5.5 处理 CSV 中的空值
- 5.6 处理大文件 (流式读取)
- 5.7 导入前清理旧数据/表
- 综合代码示例：完整导入流程
- 总结
1. 引言：CSV 与 SQLite 的桥梁
CSV (Comma Separated Values) 是一种常用的纯文本文件格式，用于存储表格数据。它简单、易读，并且几乎所有数据处理工具都支持。SQLite 是一种轻量级、无服务器的嵌入式数据库，非常适合本地应用程序、小型项目或作为数据交换的中间存储。
1.1 为什么需要导入 CSV 到 SQLite？
- 结构化存储：将非结构化或半结构化的 CSV 数据转换为结构化的数据库表，便于查询、分析和管理。
- 数据整合：将来自不同 CSV 文件的数据整合到一个统一的数据库中。
- 应用程序本地存储：作为桌面应用、移动应用或小型Web应用的本地数据存储方案。
- 数据分析准备：利用 SQL 强大的查询能力对 CSV 数据进行过滤、排序、聚合。
- 版本控制与备份：数据库文件更容易进行备份和版本管理。
1.2 Python sqlite3 与 csv 模块简介
sqlite3模块：Python 标准库的一部分，提供了与 SQLite 数据库交互的接口。它遵循 Python DB-API 2.0 规范。csv模块：Python 标准库的一部分，专门用于读取和写入 CSV 文件。它能正确处理 CSV 格式的复杂性，如带引号的字段、不同分隔符等。
2. 准备工作：理解 CSV 结构与 SQLite 数据类型
2.1 CSV 文件结构概述
CSV 文件通常由以下部分组成：
- 头部 (Header Row)：第一行通常包含列的名称。
- 数据行 (Data Rows)：后续行包含实际的数据。
- 分隔符 (Delimiter)：通常是逗号 (
,)，但也可以是分号 (;)、制表符 (\t) 等。 - 引号 (Quotes)：如果字段包含分隔符或换行符，通常会用双引号 (
") 包裹。
示例 products.csv 文件：
product_id,product_name,category,price,stock_quantity,last_updated
101,Laptop Pro,Electronics,1200.50,50,2023-01-15
102,Wireless Mouse,Electronics,25.99,200,2023-01-10
103,Mechanical Keyboard,Electronics,85.00,75,2023-01-20
104,Office Chair,Furniture,"150.00",30,2023-01-18
105,Desk Lamp,Furniture,35.00,120,2023-01-12
2.2 SQLite 数据类型亲和性 (Type Affinity)
SQLite 是动态类型数据库，这意味着你可以在任何列中存储任何类型的数据。然而，它支持“类型亲和性”，这有助于更好地组织数据：
INTEGER: 用于整数值。TEXT: 用于字符串。REAL: 用于浮点数。BLOB: 用于二进制数据。NUMERIC: 可以存储任何类型的数据。在处理时，它会尝试将其转换为 INTEGER 或 REAL。
在从 CSV 导入时，所有数据都是字符串。你需要决定将它们存储为 TEXT 还是根据其内容（如数字）将其转换为 INTEGER 或 REAL。通常，将 CSV 数据导入为 TEXT 是最安全的默认选择，然后根据需要进行类型转换。
3. 核心步骤解析
3.1 创建示例 CSV 文件
我们首先需要一个 CSV 文件来测试导入功能。在实际应用中，你将直接使用现有的 CSV 文件。
3.2 读取 CSV 文件：Python csv 模块
Python 的 csv 模块提供了两种主要方式来读取 CSV 数据：
csv.reader:- 按行读取，每行作为一个列表。
- 适用于 CSV 文件没有头部或你不需要使用头部名称来引用列的情况。
- 你必须通过索引访问列 (例如
row[0],row[1])。
csv.DictReader:- 自动将 CSV 文件的第一行作为字段名，然后将每行数据读取为一个字典，其中键是字段名。
- 更推荐，因为它使得通过列名引用数据变得更容易，尤其是在列顺序可能变化或你只想导入部分列时。
示例使用 csv.DictReader:
import csv
def read_csv_data(csv_filepath):
with open(csv_filepath, 'r', newline='', encoding='utf-8') as csvfile:
reader = csv.DictReader(csvfile)
headers = reader.fieldnames # 获取CSV头部
data = []
for row in reader:
data.append(row)
return headers, data
3.3 创建 SQLite 表结构
这是导入过程中最关键的步骤之一，表结构需要与 CSV 数据的列数和数据类型相匹配。
方法一：手动定义表结构 (预知 schema)
如果你已经知道 CSV 文件的结构，可以直接编写 CREATE TABLE 语句。
CREATE TABLE IF NOT EXISTS products (
product_id INTEGER PRIMARY KEY,
product_name TEXT NOT NULL,
category TEXT,
price REAL,
stock_quantity INTEGER,
last_updated DATE
);
优点：明确控制数据类型和约束。
缺点：不够灵活，CSV 结构变化时需要手动修改。
方法二：动态从 CSV 头部推断表结构 (更灵活)
这是更通用的方法。通过读取 CSV 文件的头部，动态构建 CREATE TABLE 语句。对于数据类型，最安全的做法是先全部定义为 TEXT，除非你确定某个列总是数值类型。
def create_table_from_csv_headers(cursor, table_name, headers):
"""
根据CSV头部动态创建表。所有列默认为TEXT类型。
"""
columns_sql = []
for header in headers:
# 简单处理：将空格和特殊字符替换为下划线，以适应SQL列名规则
sanitized_header = header.strip().replace(' ', '_').replace('-', '_').replace('.', '_')
if not sanitized_header: # 避免空列名
continue
columns_sql.append(f"`{sanitized_header}` TEXT") # 用反引号包围，允许更多列名
if not columns_sql:
raise ValueError("No valid columns found from CSV headers.")
create_table_sql = f"CREATE TABLE IF NOT EXISTS {table_name} ({', '.join(columns_sql)});"
cursor.execute(create_table_sql)
print(f" Table '{table_name}' created or already exists.")
return [col_name.strip('`') for col_name in columns_sql] # 返回清理后的列名列表
优点：高度自动化，CSV 结构变化时无需修改代码。
缺点：所有列默认为 TEXT (除非手动添加类型推断逻辑)，可能丢失一些数据库约束信息。
3.4 将数据插入 SQLite 表
单行插入的局限性
虽然可以使用 cursor.execute() 循环逐行插入，但这效率非常低，因为每次插入都会产生数据库 I/O 开销。
# 低效的单行插入示例 (不推荐用于大量数据)
# for row in data:
# values = tuple(row.values())
# cursor.execute(insert_sql, values)
批量插入 (executemany) - 推荐
cursor.executemany() 方法用于执行批量操作，它接受一个 SQL 语句和一个包含多行数据的列表（或迭代器）。这是导入大量数据时最推荐、最高效的方式。
def insert_data_batch(cursor, table_name, columns, data):
"""
使用 executemany 批量插入数据。
"""
placeholders = ', '.join(['?' for _ in columns])
insert_sql = f"INSERT INTO {table_name} ({', '.join(f'`{col}`' for col in columns)}) VALUES ({placeholders});"
# 准备要插入的数据列表 (每个元素是一个元组，对应SQL的VALUES)
# 确保字典的values()顺序与headers的顺序一致
rows_to_insert = []
for row_dict in data:
row_values = []
for col_name in columns:
row_values.append(row_dict.get(col_name)) # 使用.get()处理可能缺失的列
rows_to_insert.append(tuple(row_values))
cursor.executemany(insert_sql, rows_to_insert)
print(f" Successfully inserted {cursor.rowcount} rows into '{table_name}'.")
事务管理的重要性
数据库操作应该在事务中进行。对于导入数据，最好在所有数据插入完成后进行一次 conn.commit()。如果发生错误，可以通过 conn.rollback() 撤销所有未提交的更改。
conn.commit()：将所有更改永久保存到数据库。conn.rollback()：撤销自上次commit()以来或连接开始以来的所有更改。
4. 分步代码实现
4.1 准备：创建示例 CSV 文件
首先，我们创建一个 products.csv 文件。
import os
import csv
import sqlite3
# --- 配置 ---
CSV_FILE = "products.csv"
DB_FILE = "mydatabase.db"
TABLE_NAME = "products"
def create_sample_csv(filename=CSV_FILE):
"""创建用于测试的CSV文件。"""
print(f"Creating sample CSV file: {filename}")
headers = ["product_id", "product_name", "category", "price", "stock_quantity", "last_updated"]
data = [
[101, "Laptop Pro", "Electronics", 1200.50, 50, "2023-01-15"],
[102, "Wireless Mouse", "Electronics", 25.99, 200, "2023-01-10"],
[103, "Mechanical Keyboard", "Electronics", 85.00, 75, "2023-01-20"],
[104, "Office Chair", "Furniture", 150.00, 30, "2023-01-18"],
[105, "Desk Lamp", "Furniture", 35.00, 120, "2023-01-12"],
[106, "Webcam", "Electronics", 49.99, 100, "2023-02-01"]
]
with open(filename, 'w', newline='', encoding='utf-8') as f:
writer = csv.writer(f)
writer.writerow(headers)
writer.writerows(data)
print(f" {len(data)} rows written to {filename}")
def cleanup(filename=CSV_FILE, db_filename=DB_FILE):
"""清理生成的CSV和DB文件。"""
print("\n--- Cleaning up ---")
if os.path.exists(filename):
os.remove(filename)
print(f" Deleted CSV file: {filename}")
if os.path.exists(db_filename):
os.remove(db_filename)
print(f" Deleted DB file: {db_filename}")
def read_data_from_csv(csv_filepath):
"""读取CSV文件，返回头部和数据列表(字典形式)。"""
data = []
headers = []
try:
with open(csv_filepath, 'r', newline='', encoding='utf-8') as csvfile:
reader = csv.DictReader(csvfile)
headers = reader.fieldnames
for row in reader:
# 简单清理头部，以匹配SQLite列名，并统一键名
cleaned_row = {col.strip().replace(' ', '_').replace('-', '_').replace('.', '_'): value for col, value in row.items()}
data.append(cleaned_row)
print(f"Read {len(data)} rows from {csv_filepath}. Headers: {headers}")
return headers, data
except FileNotFoundError:
print(f"Error: CSV file '{csv_filepath}' not found.")
return [], []
except Exception as e:
print(f"Error reading CSV file: {e}")
return [], []
4.2 方法一：使用 csv.reader 和预定义表结构
这种方法需要你预先知道表的 schema。
def import_csv_to_sqlite_method1(csv_filepath, db_filepath, table_name):
"""
方法一: 使用 csv.reader 和预定义表结构导入CSV。
"""
print(f"\n--- Method 1: Importing '{csv_filepath}' to '{table_name}' with predefined schema ---")
# 预定义的表结构 (手动定义，必须匹配CSV的列顺序和类型)
create_table_sql = f'''
CREATE TABLE IF NOT EXISTS {table_name} (
product_id INTEGER PRIMARY KEY,
product_name TEXT NOT NULL,
category TEXT,
price REAL,
stock_quantity INTEGER,
last_updated DATE
);
'''
# 插入语句，参数占位符与表结构顺序一致
insert_sql = f'''
INSERT INTO {table_name} (product_id, product_name, category, price, stock_quantity, last_updated)
VALUES (?, ?, ?, ?, ?, ?);
'''
try:
with sqlite3.connect(db_filepath) as conn:
cursor = conn.cursor()
# 清理旧表 (如果存在)
cursor.execute(f"DROP TABLE IF EXISTS {table_name};")
conn.commit()
print(f" Dropped existing table '{table_name}'.")
# 创建表
cursor.execute(create_table_sql)
conn.commit()
print(f" Table '{table_name}' created.")
# 读取CSV并准备数据
rows_to_insert = []
with open(csv_filepath, 'r', newline='', encoding='utf-8') as csvfile:
reader = csv.reader(csvfile)
next(reader) # 跳过头部行
for row in reader:
# 在这里进行类型转换 (如果需要)
# 注意：CSV读取的都是字符串，根据SQLITE列类型手动转换
processed_row = [
int(row[0]) if row[0] else None, # product_id
row[1], # product_name
row[2], # category
float(row[3]) if row[3] else None, # price
int(row[4]) if row[4] else None, # stock_quantity
row[5] # last_updated
]
rows_to_insert.append(processed_row)
# 批量插入数据
if rows_to_insert:
cursor.executemany(insert_sql, rows_to_insert)
conn.commit()
print(f" Successfully imported {len(rows_to_insert)} rows into '{table_name}'.")
else:
print(" No data to import.")
except FileNotFoundError:
print(f"Error: CSV file '{csv_filepath}' not found.")
except sqlite3.Error as e:
print(f"Database error during import: {e}")
except ValueError as e:
print(f"Data type conversion error: {e}. Check CSV data and table schema.")
except Exception as e:
print(f"An unexpected error occurred: {e}")
# Example usage:
# cleanup()
# create_sample_csv()
# import_csv_to_sqlite_method1(CSV_FILE, DB_FILE, TABLE_NAME)
4.3 方法二：使用 csv.DictReader 和动态表结构 (推荐且更通用)
这种方法更灵活，能够适应 CSV 列的顺序变化，甚至可以从不同的 CSV 文件导入。
def import_csv_to_sqlite_method2(csv_filepath, db_filepath, table_name):
"""
方法二: 使用 csv.DictReader 和动态推断表结构导入CSV。
"""
print(f"\n--- Method 2: Importing '{csv_filepath}' to '{table_name}' with dynamic schema ---")
try:
# 1. 读取CSV数据
headers, data = read_data_from_csv(csv_filepath)
if not headers or not data:
print(" No data or headers found in CSV. Aborting import.")
return
with sqlite3.connect(db_filepath) as conn:
cursor = conn.cursor()
# 清理旧表 (如果存在)
cursor.execute(f"DROP TABLE IF EXISTS {table_name};")
conn.commit()
print(f" Dropped existing table '{table_name}'.")
# 2. 动态创建表
# sanitized_headers = [col.strip().replace(' ', '_').replace('-', '_').replace('.', '_') for col in headers]
# 确保使用字典键作为列名，因为read_data_from_csv已经清理过键名
cleaned_headers_from_data = list(data[0].keys())
create_table_from_csv_headers(cursor, table_name, cleaned_headers_from_data)
conn.commit()
# 3. 准备数据进行批量插入
placeholders = ', '.join(['?' for _ in cleaned_headers_from_data])
insert_sql = f"INSERT INTO {table_name} ({', '.join(f'`{col}`' for col in cleaned_headers_from_data)}) VALUES ({placeholders});"
rows_to_insert = []
for row_dict in data:
# 再次强调：这里只是获取值，并没有做显式类型转换。
# SQLite会根据其亲和性进行隐式转换（如"123"会存为INTEGER）
# 如果需要严格控制类型，可以在这里对row_dict[col]进行转换，
# 但需要判断CSV值是否为空字符串等
row_values = [row_dict.get(col_name) for col_name in cleaned_headers_from_data]
rows_to_insert.append(tuple(row_values))
# 4. 批量插入数据
if rows_to_insert:
cursor.executemany(insert_sql, rows_to_insert)
conn.commit()
print(f" Successfully imported {cursor.rowcount} rows into '{table_name}'.")
else:
print(" No data to import.")
except FileNotFoundError:
print(f"Error: CSV file '{csv_filepath}' not found.")
except sqlite3.Error as e:
print(f"Database error during import: {e}")
except Exception as e:
print(f"An unexpected error occurred: {e}")
# Example usage:
# cleanup()
# create_sample_csv()
# import_csv_to_sqlite_method2(CSV_FILE, DB_FILE, TABLE_NAME)
5. 最佳实践与注意事项
5.1 错误处理 (文件、数据库、数据)
- 文件错误：使用
try-except FileNotFoundError处理 CSV 文件不存在的情况。 - 数据库错误：使用
try-except sqlite3.Error捕获数据库操作可能发生的错误（如表名/列名错误、约束冲突等）。 - 数据类型转换错误：如果在导入前进行显式类型转换（如
int()或float()），请使用try-except ValueError来处理非法的数值格式。
5.2 数据类型转换 (字符串到数值/日期)
CSV 中的所有数据都是字符串。SQLite 会尝试根据列的类型亲和性将字符串转换为数值类型。例如，如果 price 列定义为 REAL 且 CSV 值为 “1200.50”，SQLite 会自动将其存储为浮点数。
- 显式转换：如果你需要更严格的控制（例如，确保日期格式正确，或在转换失败时抛出错误），应在 Python 代码中进行显式转换。
# 示例：将字符串转换为整数，处理空字符串 value = row_dict.get('stock_quantity', '') stock = int(value) if value.strip() else None - 日期/时间：SQLite 没有专门的日期/时间类型。通常，日期/时间以
TEXT(ISO8601 字符串格式，如 ‘YYYY-MM-DD HH:MM:SS.SSS’)、REAL(Julian day numbers) 或INTEGER(Unix epoch time) 存储。推荐使用TEXT格式。
5.3 性能优化：批量插入和提交频率
executemany()：始终使用executemany()进行批量插入，而不是在循环中调用execute()。这大大减少了 Python 和数据库之间的通信开销。- 单次
commit()：对于大多数中小型 CSV 文件，最好在所有数据插入完成后一次性调用conn.commit()。在一个大事务中执行所有操作比多次小事务更高效。 - 分批提交：对于非常大的 CSV 文件（例如，数百万行），将数据分成批次（例如，每 10000 行提交一次）可能有助于减少内存消耗和避免长时间的事务锁定。
5.4 编码问题 (encoding 参数)
CSV 文件可能有不同的编码（如 UTF-8, Latin-1, GBK 等）。在打开 CSV 文件时，务必使用正确的 encoding 参数。encoding='utf-8' 是最常见的选择。
with open(csv_filepath, 'r', newline='', encoding='utf-8') as csvfile:
# ...
5.5 处理 CSV 中的空值
CSV 文件中的空字段在 Python 中通常被读取为空字符串 ''。在插入数据库时：
- 如果数据库列允许
NULL，将空字符串转换为None(None会被 SQLite 存储为NULL) 是一个好习惯。 - 如果数据库列定义为
NOT NULL，那么空字符串在插入时可能会导致IntegrityError。你需要决定是提供默认值还是跳过该行。
5.6 处理大文件 (流式读取)
对于内存有限或文件极大的情况，避免一次性将所有 CSV 数据加载到内存中。csv.DictReader 和 csv.reader 本身就是迭代器，可以实现流式读取。结合 executemany 和分批提交，可以高效处理大文件。
# 伪代码：分批处理大文件
# batch_size = 10000
# rows_to_insert = []
# for i, row_dict in enumerate(reader):
# # ... 处理 row_dict ...
# rows_to_insert.append(processed_row_tuple)
# if (i + 1) % batch_size == 0:
# cursor.executemany(insert_sql, rows_to_insert)
# conn.commit()
# rows_to_insert = []
# if rows_to_insert: # 插入剩余的行
# cursor.executemany(insert_sql, rows_to_insert)
# conn.commit()
5.7 导入前清理旧数据/表
在重新导入相同数据或不同版本数据时，你可能希望在导入前清空表或删除并重建表：
DROP TABLE IF EXISTS your_table_name;(删除整个表)DELETE FROM your_table_name;(清空表数据，保留表结构)
6. 综合代码示例：完整导入流程
这个示例将展示使用 csv.DictReader 和动态表结构的方法，并包含错误处理。
import os
import csv
import sqlite3
import datetime
# --- Configuration ---
CSV_FILE = "products_data.csv"
DB_FILE = "product_catalog.db"
TABLE_NAME = "product_items"
# --- 1. Helper Functions ---
def create_sample_csv_data(filename=CSV_FILE):
"""Generates a sample CSV file for demonstration."""
print(f"\n--- Creating sample CSV file: {filename} ---")
headers = ["product_id", "product_name", "category", "price", "stock_quantity", "last_updated", "description"]
data = [
[101, "Laptop Pro X1", "Electronics", 1200.50, 50, "2023-01-15", "High performance laptop"],
[102, "Wireless Mouse Elite", "Electronics", 25.99, 200, "2023-01-10", "Ergonomic wireless mouse"],
[103, "Mechanical Keyboard RGB", "Electronics", 85.00, 75, "2023-01-20", "RGB backlit gaming keyboard"],
[104, "Ergonomic Office Chair", "Furniture", 150.00, 30, "2023-01-18", "Adjustable office chair for comfort"],
[105, "Modern Desk Lamp", "Furniture", 35.00, 120, "2023-01-12", "Sleek design with multiple light modes"],
[106, "HD Webcam 1080p", "Electronics", 49.99, 100, "2023-02-01", "Full HD video conferencing camera"],
[107, "Smart Speaker", "Audio", 79.99, 80, "2023-02-05", "Voice-controlled smart home assistant"],
[108, "Bluetooth Headset", "Audio", 59.99, 150, "2023-01-28", "Noise-cancelling Bluetooth headphones"],
[109, "Portable SSD 1TB", "Storage", 110.00, 40, "2023-02-10", ""], # Empty description
[110, "USB C Hub", "Accessories", 19.99, 300, "2023-01-05", "Multiport adapter for USB-C devices"]
]
with open(filename, 'w', newline='', encoding='utf-8') as f:
writer = csv.writer(f)
writer.writerow(headers)
writer.writerows(data)
print(f" {len(data)} rows written to {filename}")
def cleanup_files(csv_filename=CSV_FILE, db_filename=DB_FILE):
"""Deletes generated CSV and DB files."""
print("\n--- Cleaning up files ---")
if os.path.exists(csv_filename):
os.remove(csv_filename)
print(f" Deleted CSV file: {csv_filename}")
if os.path.exists(db_filename):
os.remove(db_filename)
print(f" Deleted DB file: {db_filename}")
def sanitize_column_name(header):
"""Sanitizes CSV header to be a valid SQL column name."""
return header.strip().replace(' ', '_').replace('-', '_').replace('.', '_').replace('/', '_')
def read_csv_to_dicts(csv_filepath):
"""Reads a CSV file into a list of dictionaries, sanitizing column names."""
data = []
headers = []
try:
with open(csv_filepath, 'r', newline='', encoding='utf-8') as csvfile:
reader = csv.DictReader(csvfile)
# Sanitize headers immediately after reading fieldnames
headers = [sanitize_column_name(h) for h in reader.fieldnames]
# Recreate reader with sanitized fieldnames if necessary,
# or just map original row dict keys to sanitized ones
for row in reader:
cleaned_row = {sanitize_column_name(k): v for k, v in row.items()}
data.append(cleaned_row)
print(f" Read {len(data)} rows from '{csv_filepath}'. Cleaned Headers: {headers}")
return headers, data
except FileNotFoundError:
print(f"Error: CSV file '{csv_filepath}' not found.")
return [], []
except Exception as e:
print(f"Error reading CSV file: {e}")
return [], []
def create_table_dynamically(cursor, table_name, headers):
"""Creates a table in SQLite based on sanitized CSV headers."""
columns_sql = []
for header in headers:
# Default to TEXT, for product_id we can enforce INTEGER PRIMARY KEY
if header == "product_id":
columns_sql.append(f"`{header}` INTEGER PRIMARY KEY")
elif header == "price":
columns_sql.append(f"`{header}` REAL")
elif header == "stock_quantity":
columns_sql.append(f"`{header}` INTEGER")
elif header == "last_updated":
columns_sql.append(f"`{header}` TEXT") # Store dates as TEXT (ISO 8601)
else:
columns_sql.append(f"`{header}` TEXT")
if not columns_sql:
raise ValueError("No valid columns found from CSV headers for table creation.")
create_table_sql = f"CREATE TABLE IF NOT EXISTS {table_name} ({', '.join(columns_sql)});"
print(f" Generated CREATE TABLE SQL: {create_table_sql}")
cursor.execute(create_table_sql)
print(f" Table '{table_name}' created or already exists.")
def insert_data_in_batches(cursor, table_name, columns, data, batch_size=1000):
"""Inserts data into the table in specified batches."""
placeholders = ', '.join(['?' for _ in columns])
insert_sql = f"INSERT INTO {table_name} ({', '.join(f'`{col}`' for col in columns)}) VALUES ({placeholders});"
total_inserted_rows = 0
current_batch = []
print(f" Starting batch insert into '{table_name}'...")
for i, row_dict in enumerate(data):
row_values = []
for col_name in columns:
value = row_dict.get(col_name)
# Basic type conversion/handling for None
if value == '' and (col_name not in ["product_id", "price", "stock_quantity"]):
value = None # Convert empty string to None for TEXT columns that can be NULL
elif col_name == "product_id" and value:
value = int(value)
elif col_name == "price" and value:
value = float(value)
elif col_name == "stock_quantity" and value:
value = int(value)
# For last_updated, it's TEXT, so no special conversion needed unless specific date objects are desired
row_values.append(value)
current_batch.append(tuple(row_values))
if len(current_batch) >= batch_size:
cursor.executemany(insert_sql, current_batch)
total_inserted_rows += cursor.rowcount
print(f" Inserted {cursor.rowcount} rows in current batch. Total: {total_inserted_rows}")
current_batch = []
# Insert any remaining rows in the last batch
if current_batch:
cursor.executemany(insert_sql, current_batch)
total_inserted_rows += cursor.rowcount
print(f" Inserted {cursor.rowcount} rows in final batch. Total: {total_inserted_rows}")
print(f" Finished batch insert. Total rows inserted: {total_inserted_rows}.")
return total_inserted_rows
def verify_import(db_filepath, table_name):
"""Verifies the imported data by counting rows and fetching a sample."""
print(f"\n--- Verifying import into '{table_name}' ---")
try:
with sqlite3.connect(db_filepath) as conn:
cursor = conn.cursor()
cursor.execute(f"SELECT COUNT(*) FROM {table_name};")
count = cursor.fetchone()[0]
print(f" Total rows in '{table_name}': {count}")
cursor.execute(f"PRAGMA table_info({table_name});")
cols_info = cursor.fetchall()
print(f" Table schema for '{table_name}':")
for col in cols_info:
print(f" - {col[1]} ({col[2]}) NotNull={col[3]} Default={col[4]} PK={col[5]}")
cursor.execute(f"SELECT * FROM {table_name} LIMIT 3;")
sample_rows = cursor.fetchall()
if sample_rows:
print(f" Sample data from '{table_name}':")
# Get column names for header
column_names = [description[0] for description in cursor.description]
print(" " + " | ".join(column_names))
print(" " + "-" * (len(" | ".join(column_names))))
for row in sample_rows:
print(" " + " | ".join(str(item) for item in row))
else:
print(" No sample data to display.")
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 ---
def main():
cleanup_files() # Start with a clean slate
create_sample_csv_data() # Create the CSV file
print(f"\n--- Starting CSV to SQLite Import Process for '{CSV_FILE}' ---")
try:
headers, data = read_csv_to_dicts(CSV_FILE)
if not data:
print("No data to import. Exiting.")
return
with sqlite3.connect(DB_FILE) as conn:
cursor = conn.cursor()
# Dynamically create table based on cleaned headers
create_table_dynamically(cursor, TABLE_NAME, headers)
# Perform batch insertion
inserted_count = insert_data_in_batches(cursor, TABLE_NAME, headers, data, batch_size=5) # Smaller batch size for demo
conn.commit() # Commit all changes
print(f"\nTotal {inserted_count} rows committed to database.")
verify_import(DB_FILE, TABLE_NAME) # Verify the import
except Exception as e:
print(f"An error occurred during the main import process: {e}")
print("\n--- CSV to SQLite Import Process Finished ---")
# cleanup_files() # Uncomment to delete files after run
if __name__ == "__main__":
main()
7. 总结
为您详尽解析了在 Python 中如何使用 sqlite3 和 csv 模块将 CSV 文件导入 SQLite 数据库。
核心要点回顾:
- 读取 CSV：推荐使用
csv.DictReader以字典形式读取数据，便于通过列名访问。 - 动态创建表：通过读取 CSV 头部（并进行清理/规范化），可以动态构建
CREATE TABLE语句，提高代码灵活性和复用性。 - 批量插入：使用
cursor.executemany()是导入大量数据时最高效的方法，结合事务管理（conn.commit()）。 - 数据类型与转换：SQLite 具有类型亲和性，但 CSV 数据本质是字符串。在导入时可以进行显式类型转换，或依赖 SQLite 的隐式转换（但需了解其行为）。
- 错误处理：始终使用
try-except块处理文件操作、数据库操作和数据转换可能发生的错误。 - 最佳实践：注意 CSV 编码、空值处理、性能优化（批量提交）、以及导入前清理旧数据。
通过遵循本指南，您将能够构建一个强大、灵活且高效的 Python 脚本，以满足将 CSV 数据导入 SQLite 数据库的各种需求。
更多推荐
23
0- 0
已为社区贡献114条内容
所有评论(0)