如何通过代码及Python代码检查数据库中Demo Data是否已加载

既然你明确需要Python代码方案来验证Demo数据是否已加载到数据库，下面针对常见的数据库类型给出具体实现，核心思路分为两种：检查目标表的记录数，或者验证是否存在标志性的Demo数据记录（后者比单纯行数检查更准确，能避免空表或非Demo测试数据的干扰）。

1. SQLite 实现示例 #

SQLite是文件型数据库，无需额外服务，适合本地测试场景：

import sqlite3

def check_demo_data_sqlite(db_path, target_table, demo_identifier=None):
    conn = None
    try:
        # 连接数据库文件
        conn = sqlite3.connect(db_path)
        cursor = conn.cursor()

        if demo_identifier:
            # 检查是否存在特定Demo数据（比如users表中name为"Demo Admin"的记录）
            # 用参数化查询避免SQL注入风险
            query = f"SELECT COUNT(*) FROM {target_table} WHERE {demo_identifier[0]} = ?"
            cursor.execute(query, (demo_identifier[1],))
            count = cursor.fetchone()[0]
            if count > 0:
                print("✅ Demo数据已加载：检测到标志性Demo记录")
                return True
            else:
                print("❌ Demo数据未加载：未找到标志性Demo记录")
                return False
        else:
            # 仅检查目标表是否有数据（行数>0）
            query = f"SELECT COUNT(*) FROM {target_table}"
            cursor.execute(query)
            count = cursor.fetchone()[0]
            if count > 0:
                print(f"✅ Demo数据已加载：{target_table}表中有{count}条记录")
                return True
            else:
                print(f"❌ Demo数据未加载：{target_table}表为空")
                return False
    except sqlite3.Error as e:
        print(f"数据库查询出错：{e}")
        return False
    finally:
        # 确保关闭连接，释放资源
        if conn:
            conn.close()

# 使用示例
# 方式1：检查特定Demo记录
check_demo_data_sqlite("my_local_db.db", "users", ("name", "Demo Admin"))
# 方式2：仅检查表是否有数据
# check_demo_data_sqlite("my_local_db.db", "products")

2. PostgreSQL 实现示例（使用psycopg2） #

先安装依赖：pip install psycopg2-binary

import psycopg2
from psycopg2 import OperationalError

def check_demo_data_postgres(db_config, target_table, demo_identifier=None):
    conn = None
    try:
        # 连接PostgreSQL数据库
        conn = psycopg2.connect(
            dbname=db_config["dbname"],
            user=db_config["user"],
            password=db_config["password"],
            host=db_config["host"],
            port=db_config["port"]
        )
        cursor = conn.cursor()

        if demo_identifier:
            # 检查特定Demo记录（比如orders表中order_id为"DEMO_001"的订单）
            query = f"SELECT COUNT(*) FROM {target_table} WHERE {demo_identifier[0]} = %s"
            cursor.execute(query, (demo_identifier[1],))
            count = cursor.fetchone()[0]
            if count > 0:
                print("✅ Demo数据已加载：检测到标志性Demo记录")
                return True
            else:
                print("❌ Demo数据未加载：未找到标志性Demo记录")
                return False
        else:
            # 检查表行数
            query = f"SELECT COUNT(*) FROM {target_table}"
            cursor.execute(query)
            count = cursor.fetchone()[0]
            if count > 0:
                print(f"✅ Demo数据已加载：{target_table}表中有{count}条记录")
                return True
            else:
                print(f"❌ Demo数据未加载：{target_table}表为空")
                return False
    except OperationalError as e:
        print(f"数据库连接或查询出错：{e}")
        return False
    finally:
        if conn:
            conn.close()

# 使用示例
db_config = {
    "dbname": "my_pg_db",
    "user": "postgres",
    "password": "my_pw",
    "host": "localhost",
    "port": "5432"
}
check_demo_data_postgres(db_config, "orders", ("order_id", "DEMO_001"))

3. MySQL 实现示例（使用mysql-connector-python） #

先安装依赖：pip install mysql-connector-python

import mysql.connector
from mysql.connector import Error

def check_demo_data_mysql(db_config, target_table, demo_identifier=None):
    conn = None
    try:
        # 连接MySQL数据库
        conn = mysql.connector.connect(
            database=db_config["database"],
            user=db_config["user"],
            password=db_config["password"],
            host=db_config["host"],
            port=db_config["port"]
        )
        if conn.is_connected():
            cursor = conn.cursor()

            if demo_identifier:
                # 检查特定Demo记录
                query = f"SELECT COUNT(*) FROM {target_table} WHERE {demo_identifier[0]} = %s"
                cursor.execute(query, (demo_identifier[1],))
                count = cursor.fetchone()[0]
                if count > 0:
                    print("✅ Demo数据已加载：检测到标志性Demo记录")
                    return True
                else:
                    print("❌ Demo数据未加载：未找到标志性Demo记录")
                    return False
            else:
                # 检查表行数
                query = f"SELECT COUNT(*) FROM {target_table}"
                cursor.execute(query)
                count = cursor.fetchone()[0]
                if count > 0:
                    print(f"✅ Demo数据已加载：{target_table}表中有{count}条记录")
                    return True
                else:
                    print(f"❌ Demo数据未加载：{target_table}表为空")
                    return False
    except Error as e:
        print(f"数据库连接或查询出错：{e}")
        return False
    finally:
        if conn and conn.is_connected():
            cursor.close()
            conn.close()

# 使用示例
db_config = {
    "database": "my_mysql_db",
    "user": "root",
    "password": "my_pw",
    "host": "localhost",
    "port": "3306"
}
check_demo_data_mysql(db_config, "products")

关键注意事项 #

• 优先用标志性数据检查：单纯行数检查可能误判（比如初始化后的空表或测试数据），用Demo数据独有的字段值（如特定ID、名称）更精准。
• 异常处理：代码中加入了数据库连接和查询的异常捕获，避免程序直接崩溃。
• 参数化查询：所有变量查询都用参数化方式，避免SQL注入风险（即使是内部工具，养成好习惯很重要）。
• 资源释放：在finally块中确保关闭数据库连接，避免资源泄漏。

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