如何用脚本批量创建SQLite表？

本文目录导读：

1. 使用SQL脚本文件（最基础）
2. Python脚本（推荐）
3. 从配置文件生成（可扩展性强）
4. 从CSV文件批量创建表
5. 批量创建并初始化的完整示例
6. 选择建议

可以使用脚本批量创建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表,你可以根据项目需求选择最合适的方案。