如何用Python检查SQLite数据库中表是否存在

import sqlite3
import os

DB_FILE = "my_database.db"

def connect_db():
    """建立数据库连接"""
    try:
        conn = sqlite3.connect(DB_FILE)
        return conn
    except sqlite3.Error as e:
        print(f"连接数据库失败: {e}")
        return None

def close_db(conn):
    """关闭数据库连接"""
    if conn:
        conn.close()

# 示例：创建测试数据库和表
def setup_test_db():
    if os.path.exists(DB_FILE):
        os.remove(DB_FILE) # 清理旧的测试文件
    conn = connect_db()
    if conn:
        cursor = conn.cursor()
        cursor.execute('''
            CREATE TABLE IF NOT EXISTS users (
                id INTEGER PRIMARY KEY,
                name TEXT NOT NULL,
                email TEXT UNIQUE
            );
        ''')
        conn.commit()
        print(f"数据库 '{DB_FILE}' 已创建，并创建了 'users' 表。")
        close_db(conn)

def cleanup_test_db():
    """删除测试数据库文件"""
    if os.path.exists(DB_FILE):
        os.remove(DB_FILE)
        print(f"数据库 '{DB_FILE}' 已删除。")

# setup_test_db() # 调用以创建测试数据库
# cleanup_test_db() # 调用以删除测试数据库

SELECT name FROM sqlite_master WHERE type='table' AND name='{table_name}';

def check_table_exists_sqlite_master(db_file, table_name):
    """
    方法一：通过查询 sqlite_master 表来检查表是否存在。
    这是最推荐和最安全的方法，因为它使用参数化查询来处理表名。
    """
    conn = None
    try:
        conn = sqlite3.connect(db_file)
        cursor = conn.cursor()
        # 使用参数化查询，防止 SQL 注入
        query = "SELECT name FROM sqlite_master WHERE type='table' AND name=?;"
        cursor.execute(query, (table_name,)) # 注意 table_name 是元组
        result = cursor.fetchone()
        return result is not None
    except sqlite3.Error as e:
        print(f"检查表 '{table_name}' 存在性时发生 SQLite 错误: {e}")
        return False
    finally:
        if conn:
            conn.close()

def check_table_exists_pragma_table_info(db_file, table_name):
    """
    方法二：通过 PRAGMA table_info() 检查表是否存在。
    此方法对表名使用字符串格式化，因此需要谨慎处理用户输入。
    """
    conn = None
    try:
        conn = sqlite3.connect(db_file)
        cursor = conn.cursor()
        # 注意：PRAGMA 语句通常不支持对表名进行参数化查询，
        # 因此这里使用 f-string 格式化。如果 table_name 来自用户输入，
        # 则存在 SQL 注入风险，需要额外的验证或白名单机制。
        cursor.execute(f"PRAGMA table_info('{table_name}');")
        result = cursor.fetchall()
        return len(result) > 0 # 如果有列信息返回，则表存在
    except sqlite3.Error as e:
        print(f"检查表 '{table_name}' 存在性时发生 SQLite 错误: {e}")
        return False
    finally:
        if conn:
            conn.close()

def check_table_exists_try_select(db_file, table_name):
    """
    方法三：尝试从表中查询数据并捕获 OperationalError 异常。
    此方法使用异常处理进行流程控制，且对表名使用字符串格式化。
    """
    conn = None
    try:
        conn = sqlite3.connect(db_file)
        cursor = conn.cursor()
        # 同样，表名作为 SQL 结构的一部分，这里使用 f-string 格式化。
        # 存在 SQL 注入风险，需要额外的验证。
        cursor.execute(f"SELECT 1 FROM {table_name} LIMIT 1;")
        return True # 如果执行成功，则表存在
    except sqlite3.OperationalError:
        return False # 捕获到 OperationalError，说明表不存在
    except sqlite3.Error as e:
        print(f"检查表 '{table_name}' 存在性时发生意外的 SQLite 错误: {e}")
        return False
    finally:
        if conn:
            conn.close()

def check_table_exists_with_context(db_file, table_name):
    """使用 with 语句管理连接，并使用推荐的 sqlite_master 方法。"""
    try:
        with sqlite3.connect(db_file) as conn: # 连接将自动关闭
            cursor = conn.cursor()
            query = "SELECT name FROM sqlite_master WHERE type='table' AND name=?;"
            cursor.execute(query, (table_name,))
            result = cursor.fetchone()
            return result is not None
    except sqlite3.Error as e:
        print(f"检查表 '{table_name}' 存在性时发生 SQLite 错误: {e}")
        return False

import sqlite3
import os

DB_FILE = "my_database.db"

def setup_database_for_testing():
    """
    创建测试数据库，包含一个 'users' 表和一个大小写敏感的 '"Products"' 表，
    以及一个不存在的 'orders' 表。
    """
    if os.path.exists(DB_FILE):
        os.remove(DB_FILE)
    try:
        with sqlite3.connect(DB_FILE) as conn:
            cursor = conn.cursor()
            # 创建一个普通表 (大小写不敏感)
            cursor.execute('''
                CREATE TABLE users (
                    id INTEGER PRIMARY KEY,
                    name TEXT NOT NULL
                );
            ''')
            # 创建一个大小写敏感的表 (注意双引号)
            cursor.execute('''
                CREATE TABLE "Products" (
                    item_id INTEGER PRIMARY KEY,
                    item_name TEXT NOT NULL
                );
            ''')
            conn.commit()
            print(f"测试数据库 '{DB_FILE}' 已创建，包含 'users' 和 'Products' 表。")
    except sqlite3.Error as e:
        print(f"Error setting up test database: {e}")

def cleanup_database_after_testing():
    """删除测试数据库文件。"""
    if os.path.exists(DB_FILE):
        os.remove(DB_FILE)
        print(f"测试数据库 '{DB_FILE}' 已删除。")

def check_table_exists_master_method(table_name):
    """方法一：查询 sqlite_master 表（推荐）。"""
    try:
        with sqlite3.connect(DB_FILE) as conn:
            cursor = conn.cursor()
            query = "SELECT name FROM sqlite_master WHERE type='table' AND name=?;"
            cursor.execute(query, (table_name,))
            return cursor.fetchone() is not None
    except sqlite3.Error as e:
        print(f"SQLite error for '{table_name}' with master method: {e}")
        return False

def check_table_exists_pragma_method(table_name):
    """方法二：使用 PRAGMA table_info()（注意 SQL 注入风险）。"""
    try:
        with sqlite3.connect(DB_FILE) as conn:
            cursor = conn.cursor()
            # !! SQL 注入风险：表名未经参数化，来自用户输入时需严格验证 !!
            cursor.execute(f"PRAGMA table_info('{table_name}');")
            return len(cursor.fetchall()) > 0
    except sqlite3.Error as e:
        print(f"SQLite error for '{table_name}' with pragma method: {e}")
        return False

def check_table_exists_try_select_method(table_name):
    """方法三：尝试查询并捕获异常（注意 SQL 注入风险和流程控制）。"""
    try:
        with sqlite3.connect(DB_FILE) as conn:
            cursor = conn.cursor()
            # !! SQL 注入风险：表名未经参数化，来自用户输入时需严格验证 !!
            cursor.execute(f"SELECT 1 FROM {table_name} LIMIT 1;")
            return True
    except sqlite3.OperationalError:
        return False
    except sqlite3.Error as e:
        print(f"SQLite error for '{table_name}' with try-select method: {e}")
        return False

def main():
    setup_database_for_testing()

    print("\n--- 检查 'users' 表 (存在，大小写不敏感) ---")
    print(f"sqlite_master: {check_table_exists_master_method('users')}")
    print(f"PRAGMA: {check_table_exists_pragma_method('users')}")
    print(f"try-select: {check_table_exists_try_select_method('users')}")
    print(f"sqlite_master (大写): {check_table_exists_master_method('USERS')}") # SQLite 默认不区分大小写
    print(f"PRAGMA (大写): {check_table_exists_pragma_method('USERS')}")
    print(f"try-select (大写): {check_table_exists_try_select_method('USERS')}")


    print("\n--- 检查 'Products' 表 (存在，大小写敏感) ---")
    print(f"sqlite_master: {check_table_exists_master_method('Products')}")
    print(f"PRAGMA: {check_table_exists_pragma_method('Products')}")
    print(f"try-select: {check_table_exists_try_select_method('Products')}")
    print(f"sqlite_master (小写): {check_table_exists_master_method('products')}") # 应该为 False
    print(f"PRAGMA (小写): {check_table_exists_pragma_method('products')}")
    print(f"try-select (小写): {check_table_exists_try_select_method('products')}")


    print("\n--- 检查 'orders' 表 (不存在) ---")
    print(f"sqlite_master: {check_table_exists_master_method('orders')}")
    print(f"PRAGMA: {check_table_exists_pragma_method('orders')}")
    print(f"try-select: {check_table_exists_try_select_method('orders')}")

    cleanup_database_after_testing()

if __name__ == "__main__":
    main()