本文目录导读：
可以使用脚本批量创建SQLite表,以下是几种常见的方法：
使用SQL脚本文件（最基础）
创建一个SQL文件,包含所有表的CREATE语句：
create_tables.sql
CREATE TABLE IF NOT EXISTS users (
id INTEGER PRIMARY KEY AUTOINCREMENT,
name TEXT NOT NULL,
email TEXT UNIQUE,
created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP
);
CREATE TABLE IF NOT EXISTS orders (
id INTEGER PRIMARY KEY AUTOINCREMENT,
user_id INTEGER,
product TEXT NOT NULL,
amount REAL,
FOREIGN KEY (user_id) REFERENCES users(id)
);
CREATE TABLE IF NOT EXISTS products (
id INTEGER PRIMARY KEY AUTOINCREMENT,
name TEXT NOT NULL,
price REAL,
stock INTEGER DEFAULT 0
);
然后执行：
sqlite3 database.db < create_tables.sql
Python脚本（推荐）
import sqlite3
def create_tables():
conn = sqlite3.connect('database.db')
cursor = conn.cursor()
tables_sql = [
"""
CREATE TABLE IF NOT EXISTS users (
id INTEGER PRIMARY KEY AUTOINCREMENT,
name TEXT NOT NULL,
email TEXT UNIQUE,
created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP
)
""",
"""
CREATE TABLE IF NOT EXISTS orders (
id INTEGER PRIMARY KEY AUTOINCREMENT,
user_id INTEGER,
product TEXT NOT NULL,
amount REAL,
FOREIGN KEY (user_id) REFERENCES users(id)
)
""",
"""
CREATE TABLE IF NOT EXISTS products (
id INTEGER PRIMARY KEY AUTOINCREMENT,
name TEXT NOT NULL,
price REAL,
stock INTEGER DEFAULT 0
)
"""
]
for sql in tables_sql:
cursor.execute(sql)
print(f"Table created successfully")
conn.commit()
conn.close()
print("All tables created!")
if __name__ == "__main__":
create_tables()
从配置文件生成（可扩展性强）
config.json
{
"tables": {
"users": {
"columns": {
"id": "INTEGER PRIMARY KEY AUTOINCREMENT",
"name": "TEXT NOT NULL",
"email": "TEXT UNIQUE",
"created_at": "TIMESTAMP DEFAULT CURRENT_TIMESTAMP"
}
},
"orders": {
"columns": {
"id": "INTEGER PRIMARY KEY AUTOINCREMENT",
"user_id": "INTEGER",
"product": "TEXT NOT NULL",
"amount": "REAL"
},
"foreign_keys": [
"FOREIGN KEY (user_id) REFERENCES users(id)"
]
},
"products": {
"columns": {
"id": "INTEGER PRIMARY KEY AUTOINCREMENT",
"name": "TEXT NOT NULL",
"price": "REAL",
"stock": "INTEGER DEFAULT 0"
}
}
}
}
generate_tables.py
import json
import sqlite3
def generate_create_sql(table_name, table_config):
columns = []
for col_name, col_type in table_config['columns'].items():
columns.append(f"{col_name} {col_type}")
if 'foreign_keys' in table_config:
columns.extend(table_config['foreign_keys'])
columns_str = ",\n ".join(columns)
return f"CREATE TABLE IF NOT EXISTS {table_name} (\n {columns_str}\n)"
def create_tables_from_config(config_file):
with open(config_file, 'r') as f:
config = json.load(f)
conn = sqlite3.connect('database.db')
cursor = conn.cursor()
for table_name, table_config in config['tables'].items():
sql = generate_create_sql(table_name, table_config)
cursor.execute(sql)
print(f"Created table: {table_name}")
conn.commit()
conn.close()
print("All tables created from configuration!")
if __name__ == "__main__":
create_tables_from_config('config.json')
从CSV文件批量创建表
import sqlite3
import csv
import os
def create_tables_from_csv(csv_folder):
conn = sqlite3.connect('database.db')
cursor = conn.cursor()
for filename in os.listdir(csv_folder):
if filename.endswith('.csv'):
table_name = filename.replace('.csv', '')
with open(os.path.join(csv_folder, filename), 'r') as f:
reader = csv.reader(f)
headers = next(reader)
# 创建列定义（所有列默认为TEXT类型）
columns = []
for header in headers:
columns.append(f"{header} TEXT")
columns_str = ",\n ".join(columns)
sql = f"CREATE TABLE IF NOT EXISTS {table_name} (\n {columns_str}\n)"
cursor.execute(sql)
print(f"Created table: {table_name}")
conn.commit()
conn.close()
if __name__ == "__main__":
create_tables_from_csv('./csv_files')
批量创建并初始化的完整示例
import sqlite3
from typing import Dict, List
class SQLiteTableManager:
def __init__(self, db_path: str):
self.db_path = db_path
self.conn = None
self.cursor = None
def connect(self):
self.conn = sqlite3.connect(self.db_path)
self.cursor = self.conn.cursor()
def close(self):
if self.conn:
self.conn.close()
def create_tables(self, tables: Dict[str, Dict]) -> List[str]:
"""
tables格式:
{
'table_name': {
'columns': { 'col_name': 'col_type', ... },
'foreign_keys': ['FOREIGN KEY ...', ...] # 可选
}
}
"""
created_tables = []
for table_name, table_config in tables.items():
try:
columns = []
for col_name, col_type in table_config['columns'].items():
columns.append(f"{col_name} {col_type}")
if 'foreign_keys' in table_config:
columns.extend(table_config['foreign_keys'])
columns_str = ",\n ".join(columns)
sql = f"CREATE TABLE IF NOT EXISTS {table_name} (\n {columns_str}\n)"
self.cursor.execute(sql)
created_tables.append(table_name)
print(f"✓ Created table: {table_name}")
except Exception as e:
print(f"✗ Failed to create table {table_name}: {e}")
self.conn.commit()
return created_tables
def table_exists(self, table_name: str) -> bool:
self.cursor.execute(
"SELECT name FROM sqlite_master WHERE type='table' AND name=?",
(table_name,)
)
return self.cursor.fetchone() is not None
# 使用示例
if __name__ == "__main__":
tables_config = {
'users': {
'columns': {
'id': 'INTEGER PRIMARY KEY AUTOINCREMENT',
'username': 'TEXT NOT NULL UNIQUE',
'email': 'TEXT NOT NULL',
'password_hash': 'TEXT NOT NULL',
'created_at': 'TIMESTAMP DEFAULT CURRENT_TIMESTAMP'
}
},
'posts': {
'columns': {
'id': 'INTEGER PRIMARY KEY AUTOINCREMENT',
'user_id': 'INTEGER NOT NULL',
'title': 'TEXT NOT NULL',
'content': 'TEXT',
'created_at': 'TIMESTAMP DEFAULT CURRENT_TIMESTAMP'
},
'foreign_keys': [
'FOREIGN KEY (user_id) REFERENCES users(id)'
]
},
'comments': {
'columns': {
'id': 'INTEGER PRIMARY KEY AUTOINCREMENT',
'post_id': 'INTEGER NOT NULL',
'user_id': 'INTEGER NOT NULL',
'content': 'TEXT NOT NULL',
'created_at': 'TIMESTAMP DEFAULT CURRENT_TIMESTAMP'
},
'foreign_keys': [
'FOREIGN KEY (post_id) REFERENCES posts(id)',
'FOREIGN KEY (user_id) REFERENCES users(id)'
]
}
}
manager = SQLiteTableManager('my_database.db')
manager.connect()
created = manager.create_tables(tables_config)
print(f"\nSuccessfully created {len(created)} tables")
# 检查表是否存在
for table in ['users', 'posts', 'comments']:
exists = manager.table_exists(table)
print(f"Table '{table}' exists: {exists}")
manager.close()
选择建议
- 小型项目：直接使用SQL脚本文件
- 中型项目：使用Python脚本，易于维护和扩展
- 大型项目：从配置文件生成，便于管理和版本控制
- 数据导入：从CSV文件自动创建
这些方法都可以批量创建SQLite表,你可以根据项目需求选择最合适的方案。
