database-seeding

数据库填充

此技能可让 AI Agent 为开发、测试和预发布环境生成逼真的测试数据并插入数据库。Agent 会使用确定性生成器或 Faker 库创建可幂等执行的填充脚本，按照正确的外键顺序处理关系数据，支持特定环境的填充配置（精简的开发数据与大规模负载测试数据），并确保可以重复运行填充脚本而不会造成数据重复。

工作流

1. 分析目标架构： 检查数据库架构，确定所有表、列、数据类型、约束（NOT NULL、UNIQUE、CHECK、外键）和关系。确定满足外键依赖的正确插入顺序——必须先填充父表，再填充子表。

2. 设计数据填充策略： 根据使用场景选择合适的方法。对于可复现的测试套件，使用固定种子的确定性数据。对于逼真的开发数据，使用基于 Faker 的生成方式。对于需要逼真数据分布的预发布环境，使用经过匿名化的生产环境快照。为每张表定义数据量。

3. 生成填充脚本： 使用项目所用语言（Python、JavaScript、SQL 等）编写符合所有架构约束的数据填充脚本。使用 Faker 库或同类工具生成逼真的姓名、邮箱、地址和日期。通过生成唯一值或使用基于序列的模式来处理唯一约束。将插入操作包装在事务中，确保原子性。

4. 确保幂等性： 设计可安全重复运行的脚本。使用 INSERT ON CONFLICT DO NOTHING、UPSERT 模式或先清空再插入的策略。插入前检查是否已有数据，以避免重复运行时出现重复数据或违反约束。

5. 支持特定环境的配置： 创建不同的填充配置——本地开发使用小型数据集（每张表 10-50 条记录），集成测试使用中型数据集（1,000-10,000 条记录），性能测试使用大型数据集（100K+ 条记录）。通过环境变量或命令行参数控制配置。

6. 执行并验证： 对目标数据库运行填充脚本，验证行数是否符合预期，并通过检查所有外键是否引用现有行来确认关系完整性。记录填充结果及每张表的记录数。

支持的技术

• Python： Faker、Factory Boy、SQLAlchemy、psycopg2
• JavaScript/TypeScript： @faker-js/faker、Prisma seed、Knex seed files、TypeORM
• SQL： 原始 INSERT 语句、COPY FROM CSV
• Ruby： FactoryBot、Faker gem、Rails db:seed
• 框架： Django fixtures、Laravel seeders、Rails seeds.rb

用法

提供数据库架构（或指定迁移文件），并说明目标环境和所需数据量。Agent 会生成完整的填充脚本，遵循所有约束和关系。你可以指定数据特征（例如，“包含来自多个时区的用户”或“创建涵盖过去 12 个月的订单”）。

示例

示例 1：使用 Faker 的 Python 填充脚本

请求： 使用用户、产品和订单数据填充 PostgreSQL 数据库，用于开发。

"""seed.py — Seed development database with realistic test data."""
import random
from datetime import datetime, timedelta
from faker import Faker
import psycopg2

fake = Faker()
Faker.seed(42)  # Deterministic output for reproducibility
random.seed(42)

DB_CONFIG = {
    "host": "localhost",
    "port": 5432,
    "dbname": "dev_db",
    "user": "dev_user",
    "password": "dev_password",
}

NUM_USERS = 50
NUM_PRODUCTS = 30
NUM_ORDERS = 100


def seed():
    conn = psycopg2.connect(**DB_CONFIG)
    cur = conn.cursor()

    # Seed users
    user_ids = []
    for _ in range(NUM_USERS):
        cur.execute(
            """INSERT INTO users (email, password_hash, full_name, created_at)
               VALUES (%s, %s, %s, %s)
               ON CONFLICT (email) DO NOTHING
               RETURNING id""",
            (
                fake.unique.email(),
                fake.sha256(),
                fake.name(),
                fake.date_time_between(start_date="-2y", end_date="now"),
            ),
        )
        row = cur.fetchone()
        if row:
            user_ids.append(row[0])

    # Seed products
    product_ids = []
    for i in range(NUM_PRODUCTS):
        cur.execute(
            """INSERT INTO products (name, description, price, stock_quantity, sku)
               VALUES (%s, %s, %s, %s, %s)
               ON CONFLICT (sku) DO NOTHING
               RETURNING id""",
            (
                fake.catch_phrase(),
                fake.paragraph(nb_sentences=3),
                round(random.uniform(9.99, 499.99), 2),
                random.randint(0, 500),
                f"SKU-{i+1:05d}",
            ),
        )
        row = cur.fetchone()
        if row:
            product_ids.append(row[0])

    # Seed orders with order items
    statuses = ["pending", "confirmed", "shipped", "delivered"]
    for _ in range(NUM_ORDERS):
        user_id = random.choice(user_ids)
        status = random.choice(statuses)
        items = random.sample(product_ids, k=random.randint(1, 5))
        total = 0.0

        cur.execute(
            """INSERT INTO orders (user_id, status, total_amount, shipping_address, ordered_at)
               VALUES (%s, %s, 0, %s, %s) RETURNING id""",
            (user_id, status, fake.address(), fake.date_time_between("-1y", "now")),
        )
        order_id = cur.fetchone()[0]

        for pid in items:
            qty = random.randint(1, 4)
            price = round(random.uniform(9.99, 499.99), 2)
            total += qty * price
            cur.execute(
                """INSERT INTO order_items (order_id, product_id, quantity, unit_price)
                   VALUES (%s, %s, %s, %s)""",
                (order_id, pid, qty, price),
            )

        cur.execute(
            "UPDATE orders SET total_amount = %s WHERE id = %s", (round(total, 2), order_id)
        )

    conn.commit()
    cur.close()
    conn.close()
    print(f"Seeded {len(user_ids)} users, {len(product_ids)} products, {NUM_ORDERS} orders.")


if __name__ == "__main__":
    seed()

示例 2：包含逼真测试数据的 SQL 填充文件

请求： 为小型开发数据集创建纯 SQL 填充文件。

-- seed.sql — Idempotent seed data for local development
-- Run with: psql -U dev_user -d dev_db -f seed.sql

BEGIN;

-- Users
INSERT INTO users (id, email, password_hash, full_name, created_at) VALUES
  (1, '[email protected]',  'hash_alice',  'Alice Johnson',  '2024-03-15 09:00:00'),
  (2, '[email protected]',    'hash_bob',    'Bob Martinez',   '2024-05-20 14:30:00'),
  (3, '[email protected]',  'hash_carol',  'Carol Chen',     '2024-07-01 11:15:00'),
  (4, '[email protected]',   'hash_dave',   'Dave Okafor',    '2024-09-10 08:45:00'),
  (5, '[email protected]',    'hash_eve',    'Eve Andersson',  '2024-11-28 16:00:00')
ON CONFLICT (id) DO NOTHING;

-- Products
INSERT INTO products (id, name, description, price, stock_quantity, sku) VALUES
  (1, 'Wireless Keyboard',   'Bluetooth mechanical keyboard',  79.99,  150, 'SKU-00001'),
  (2, 'USB-C Hub',           '7-in-1 USB-C docking station',   49.99,  300, 'SKU-00002'),
  (3, 'Noise-Cancelling Headphones', 'Over-ear ANC headphones', 199.99, 75, 'SKU-00003'),
  (4, '4K Monitor',          '27-inch IPS 4K display',         399.99,  40, 'SKU-00004'),
  (5, 'Laptop Stand',        'Adjustable aluminum stand',       34.99, 200, 'SKU-00005')
ON CONFLICT (id) DO NOTHING;

-- Orders
INSERT INTO orders (id, user_id, status, total_amount, shipping_address, ordered_at) VALUES
  (1, 1, 'delivered',  129.98, '123 Oak St, Portland, OR 97201',   '2024-12-01 10:00:00'),
  (2, 2, 'shipped',    199.99, '456 Elm Ave, Austin, TX 78701',    '2025-01-05 14:20:00'),
  (3, 3, 'confirmed',  484.98, '789 Pine Rd, Seattle, WA 98101',   '2025-01-10 09:30:00'),
  (4, 1, 'pending',     49.99, '123 Oak St, Portland, OR 97201',   '2025-01-12 16:45:00')
ON CONFLICT (id) DO NOTHING;

-- Order items
INSERT INTO order_items (id, order_id, product_id, quantity, unit_price) VALUES
  (1, 1, 1, 1, 79.99),
  (2, 1, 2, 1, 49.99),
  (3, 2, 3, 1, 199.99),
  (4, 3, 4, 1, 399.99),
  (5, 3, 5, 1, 34.99),
  (6, 4, 2, 1, 49.99)
ON CONFLICT (id) DO NOTHING;

-- Reset sequences to avoid conflicts with future inserts
SELECT setval('users_id_seq',    (SELECT MAX(id) FROM users));
SELECT setval('products_id_seq', (SELECT MAX(id) FROM products));
SELECT setval('orders_id_seq',   (SELECT MAX(id) FROM orders));
SELECT setval('order_items_id_seq', (SELECT MAX(id) FROM order_items));

COMMIT;

最佳实践

• 设置固定的随机种子（例如，Faker.seed(42)），以生成确定性数据，让测试结果可复现，并使种子输出中的差异具有意义。
• 始终遵守外键顺序——先插入父行，再插入子行。编写脚本前先梳理依赖关系图，避免违反约束。
• 使用 ON CONFLICT DO NOTHING 或 UPSERT 模式，确保种子脚本具备幂等性。运行两次种子脚本应得到相同结果，而不是生成重复数据或报错。
• 按环境区分种子配置——为本地开发准备小型、快速的种子，为 CI 集成测试准备更大规模的种子，并为负载测试准备更大规模的种子。通过环境变量控制。
• 切勿向生产数据库写入测试数据。在种子脚本开头使用环境检查（例如，assert os.environ["ENV"] != "production"）作为安全防护。
• 使用显式 ID 写入种子数据后，重置自增序列，以防应用插入新行时发生主键冲突。

边缘情况

• 使用 faker 时发生唯一约束冲突： Faker 不保证大型数据集中的值唯一。使用 fake.unique.email()，或在生成的值后附加计数器以避免重复。每次测试运行之间使用 fake.unique.clear() 重置唯一性跟踪器。
• 循环外键： 当表 A 引用表 B，且表 B 也引用表 A 时，先向两个表插入外键列可为空的行，然后再通过第二轮操作更新外键值。
• 大型种子数据集与性能： 对于超过 10,000 行的种子数据，请使用批量插入方法（PostgreSQL 中的 COPY、MySQL 中的 LOAD DATA INFILE），而非逐条执行 INSERT 语句。批量加载期间禁用索引和约束，完成后再重新启用。
• 依赖时间的测试数据： 如果应用逻辑依赖日期（例如“最近 30 天内的订单”），请根据当前日期生成日期，而不要使用很快就会过时的硬编码日期。
• 写入二进制或文件数据： 对于引用上传文件或图片的表，请使用占位文件路径或较小的 base64 编码测试图片作为种子数据，而不要尝试生成完整的二进制内容。