When building and testing relational database applications, developers often face a trade-off between using empty databases (which miss edge cases and UI l
When building and testing relational database applications, developers often face a trade-off between using empty databases (which miss edge cases and UI layout issues) or copying production snapshots (which pose severe privacy/PII compliance risks and require heavy scrubbing). Simple mock generation scripts usually generate disconnected rows, breaking foreign key integrity and producing unrealistic relational graphs.
This script provides a scalable, CLI-driven database seeder that populates PostgreSQL databases with contextual data while strictly respecting foreign key constraints. By using dependency-ordered seeding graphs and batch bulk insertions, it can generate tens of thousands of realistic relational entities (such as users, categories, posts, and comments) in seconds for local development, staging, or CI/CD test pipelines.
The script operates on a dependency graph strategy to ensure relational integrity without violating foreign key constraints. Seeding proceeds top-down through database relationships: parent tables with no dependencies (e.g., users, categories) are populated first. Their primary keys are returned using PostgreSQL's RETURNING clause and buffered in memory. Child tables (e.g., posts) draw randomly from these valid parent IDs during record generation, followed by grandchild entities (comments).
To optimize performance, the script uses psycopg2.extras.execute_values for bulk vector insertions instead of executing individual INSERT statements. Configuration is driven by a YAML file that defines entity distribution counts, combined with environment variables for database credentials. The entire process runs inside a single database transaction: if any insertion step fails or a foreign key reference breaks, the script performs an automatic transaction rollback, leaving the database clean.
+-------------------+
| seed_config.yaml |
+---------+---------+
|
v
+------------------+ +----+------------------+
| Environment Vars |-->| Argument & Config |
| (DB Credentials) | | Parser |
+------------------+ +----+------------------+
|
v
+----+------------------+
| PostgreSQL Connection |
+----+------------------+
|
+------------------+------------------+
| Transaction Start |
v v
+------------------+ +------------------+
| 1. Seed Users | | Seed Categories |
| (Get User IDs) | | (Get Cat IDs) |
+--------+---------+ +--------+---------+
| |
+------------------+------------------+
|
v
+---------+---------+
| 2. Seed Posts |
| (Get Post IDs) |
+---------+---------+
|
v
+---------+---------+
| 3. Seed Comments |
+---------+---------+
|
v
+---------+---------+
| Commit / Rollback |
+-------------------+
Save the following complete script as dev_seeder.py:
#!/usr/bin/env python3
"""
Development Database Seeder with Relationships
Generates rich, context-aware synthetic seed data with valid relational integrity
for PostgreSQL databases using Faker and PyYAML.
"""
import argparse
import logging
import os
import random
import sys
from typing import Any, Dict, List, Tuple
import faker
import psycopg2
import psycopg2.extras
import yaml
# Configure structured logging output
logging.basicConfig(
level=logging.INFO,
format="%(asctime)s [%(levelname)s] %(message)s",
handlers=[logging.StreamHandler(sys.stdout)],
)
logger = logging.getLogger("dev_seeder")
def parse_args() -> argparse.Namespace:
"""Parse command line arguments."""
parser = argparse.ArgumentParser(
description="Populate PostgreSQL development database with mock relational data."
)
parser.add_argument(
"--config",
type=str,
default="seed_config.yaml",
help="Path to YAML configuration file (default: seed_config.yaml)",
)
parser.add_argument(
"--env",
type=str,
default="development",
help="Target environment name (default: development)",
)
parser.add_argument(
"--clean",
action="store_true",
help="Truncate existing tables before seeding",
)
return parser.parse_args()
def load_config(config_path: str) -> Dict[str, Any]:
"""Load and validate the YAML configuration file."""
if not os.path.exists(config_path):
logger.error("Configuration file not found: %s", config_path)
sys.exit(1)
try:
with open(config_path, "r", encoding="utf-8") as f:
config = yaml.safe_load(f)
logger.info("Successfully loaded configuration from %s", config_path)
return config
except yaml.YAMLError as err:
logger.error("Failed to parse YAML configuration: %s", err)
sys.exit(1)
def get_db_connection() -> psycopg2.extensions.connection:
"""Establish connection to PostgreSQL using environment variables with defaults."""
db_host = os.environ.get("DB_HOST", "localhost")
db_port = os.environ.get("DB_PORT", "5432")
db_name = os.environ.get("DB_NAME", "dev_db")
db_user = os.environ.get("DB_USER", "postgres")
db_password = os.environ.get("DB_PASSWORD", "postgres")
try:
conn = psycopg2.connect(
host=db_host,
port=db_port,
dbname=db_name,
user=db_user,
password=db_password,
)
conn.autocommit = False
logger.info("Connected to PostgreSQL database '%s' at %s:%s", db_name, db_host, db_port)
return conn
except psycopg2.Error as err:
logger.error("Database connection failed: %s", err)
sys.exit(1)
def create_schema_if_not_exists(cursor: psycopg2.extensions.cursor) -> None:
"""Ensure standard tables exist before seeding."""
schema_sql = """
CREATE TABLE IF NOT EXISTS users (
id SERIAL PRIMARY KEY,
username VARCHAR(50) UNIQUE NOT NULL,
email VARCHAR(100) UNIQUE NOT NULL,
full_name VARCHAR(100) NOT NULL,
created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP
);
CREATE TABLE IF NOT EXISTS categories (
id SERIAL PRIMARY KEY,
name VARCHAR(50) UNIQUE NOT NULL,
slug VARCHAR(50) UNIQUE NOT NULL,
description TEXT
);
CREATE TABLE IF NOT EXISTS posts (
id SERIAL PRIMARY KEY,
user_id INTEGER NOT NULL REFERENCES users(id) ON DELETE CASCADE,
category_id INTEGER NOT NULL REFERENCES categories(id) ON DELETE CASCADE,
title VARCHAR(200) NOT NULL,
content TEXT NOT NULL,
published BOOLEAN DEFAULT TRUE,
created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP
);
CREATE TABLE IF NOT EXISTS comments (
id SERIAL PRIMARY KEY,
post_id INTEGER NOT NULL REFERENCES posts(id) ON DELETE CASCADE,
user_id INTEGER NOT NULL REFERENCES users(id) ON DELETE CASCADE,
content TEXT NOT NULL,
created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP
);
"""
cursor.execute(schema_sql)
logger.info("Validated database schema (created tables if missing).")
def clean_tables(cursor: psycopg2.extensions.cursor) -> None:
"""Truncate all target tables in cascading order."""
logger.warning("Truncating existing data from tables...")
cursor.execute("TRUNCATE TABLE users, categories CASCADE;")
def seed_users(cursor: psycopg2.extensions.cursor, fake: faker.Faker, count: int) -> List[int]:
"""Generate and insert user records, returning their generated IDs."""
logger.info("Generating %d users...", count)
users_data: List[Tuple[str, str, str, Any]] = []
for _ in range(count):
profile = fake.simple_profile()
users_data.append((
profile["username"] + "_" + str(fake.random_int(min=100, max=9999)),
fake.unique.email(),
profile["name"],
fake.date_time_between(start_date="-2y", end_date="now")
))
insert_sql = """
INSERT INTO users (username, email, full_name, created_at)
VALUES %s
RETURNING id;
"""
# execute_values with RETURNING requires fetching results per batch
inserted_ids: List[int] = []
psycopg2.extras.execute_values(cursor, insert_sql, users_data, page_size=1000)
# Fetch returned IDs
rows = cursor.fetchall()
inserted_ids = [row[0] for row in rows]
logger.info("Successfully inserted %d users.", len(inserted_ids))
return inserted_ids
def seed_categories(cursor: psycopg2.extensions.cursor, fake: faker.Faker, count: int) -> List[int]:
"""Generate and insert category records, returning their generated IDs."""
logger.info("Generating %d categories...", count)
categories_data: List[Tuple[str, str, str]] = []
words = fake.words(nb=count + 10, unique=True)
for i in range(count):
word = str(words[i]).capitalize()
categories_data.append((
f"{word} Tech",
f"{word.lower()}-tech",
fake.sentence()
))
insert_sql = """
INSERT INTO categories (name, slug, description)
VALUES %s
RETURNING id;
"""
psycopg2.extras.execute_values(cursor, insert_sql, categories_data)
rows = cursor.fetchall()
inserted_ids = [row[0] for row in rows]
logger.info("Successfully inserted %d categories.", len(inserted_ids))
return inserted_ids
def seed_posts(
cursor: psycopg2.extensions.cursor,
fake: faker.Faker,
user_ids: List[int],
category_ids: List[int],
posts_per_user: int,
) -> List[int]:
"""Generate posts referencing existing user and category IDs."""
total_posts = len(user_ids) * posts_per_user
logger.info("Generating approx %d posts (%d per user)...", total_posts, posts_per_user)
posts_data: List[Tuple[int, int, str, str, bool, Any]] = []
for user_id in user_ids:
for _ in range(posts_per_user):
posts_data.append((
user_id,
random.choice(category_ids),
fake.sentence(nb_words=6),
fake.paragraph(nb_sentences=5),
fake.boolean(chance_of_getting_true=80),
fake.date_time_between(start_date="-1y", end_date="now")
))
insert_sql = """
INSERT INTO posts (user_id, category_id, title, content, published, created_at)
VALUES %s
RETURNING id;
"""
psycopg2.extras.execute_values(cursor, insert_sql, posts_data, page_size=1000)
rows = cursor.fetchall()
inserted_ids = [row[0] for row in rows]
logger.info("Successfully inserted %d posts.", len(inserted_ids))
return inserted_ids
def seed_comments(
cursor: psycopg2.extensions.cursor,
fake: faker.Faker,
user_ids: List[int],
post_ids: List[int],
comments_per_post: int,
) -> None:
"""Generate comments linking random users to existing posts."""
total_comments = len(post_ids) * comments_per_post
logger.info("Generating approx %d comments...", total_comments)
comments_data: List[Tuple[int, int, str, Any]] = []
for post_id in post_ids:
for _ in range(comments_per_post):
comments_data.append((
post_id,
random.choice(user_ids),
fake.paragraph(nb_sentences=2),
fake.date_time_between(start_date="-6m", end_date="now")
))
insert_sql = """
INSERT INTO comments (post_id, user_id, content, created_at)
VALUES %s;
"""
psycopg2.extras.execute_values(cursor, insert_sql, comments_data, page_size=1000)
logger.info("Successfully inserted %d comments.", len(comments_data))
def main() -> None:
"""Main execution entry point."""
args = parse_args()
logger.info("Starting database seeder [Environment: %s]", args.env)
config = load_config(args.config)
counts = config.get("counts", {})
num_users = counts.get("users", 20)
num_categories = counts.get("categories", 5)
posts_per_user = counts.get("posts_per_user", 3)
comments_per_post = counts.get("comments_per_post", 2)
fake = faker.Faker()
# Seed generator for reproducibility if specified in config
if "random_seed" in config:
faker.Faker.seed(config["random_seed"])
random.seed(config["random_seed"])
conn = get_db_connection()
try:
with conn.cursor() as cursor:
create_schema_if_not_exists(cursor)
if args.clean:
clean_tables(cursor)
# Step 1: Seed parent tables
user_ids = seed_users(cursor, fake, num_users)
category_ids = seed_categories(cursor, fake, num_categories)
# Step 2: Seed level 1 dependencies
post_ids = seed_posts(cursor, fake, user_ids, category_ids, posts_per_user)
# Step 3: Seed level 2 dependencies
seed_comments(cursor, fake, user_ids, post_ids, comments_per_post)
# Commit transaction if all operations succeed
conn.commit()
logger.info("Seeding completed successfully!")
except Exception as err:
logger.error("An error occurred during seeding. Rolling back transaction: %s", err)
conn.rollback()
sys.exit(1)
finally:
conn.close()
logger.info("Database connection closed.")
if __name__ == "__main__":
main()
The script is driven by environment variables for database connectivity and a YAML file for dataset sizing.
| Variable | Description | Default Value |
|---|---|---|
DB_HOST |
Hostname or IP address of the PostgreSQL server. | localhost |
DB_PORT |
Port number of PostgreSQL service. | 5432 |
DB_NAME |
Name of the target database. | dev_db |
DB_USER |
PostgreSQL username. | postgres |
DB_PASSWORD |
PostgreSQL user password. | postgres |
seed_config.yaml)Create a seed_config.yaml file in the working directory to define entity proportions:
# Configuration for Development Database Seeder
random_seed: 42 # Optional fixed seed for deterministic reproducible data
counts:
users: 50
categories: 8
posts_per_user: 5
comments_per_post: 4
Install the necessary dependencies before running the script:
pip install faker psycopg2-binary pyyaml
Run using defaults (seed_config.yaml and default local environment variables):
python3 dev_seeder.py
Output:
2023-10-24 10:00:00 [INFO] Starting database seeder [Environment: development]
2023-10-24 10:00:00 [INFO] Successfully loaded configuration from seed_config.yaml
2023-10-24 10:00:00 [INFO] Connected to PostgreSQL database 'dev_db' at localhost:5432
2023-10-24 10:00:00 [INFO] Validated database schema (created tables if missing).
2023-10-24 10:00:00 [INFO] Generating 50 users...
2023-10-24 10:00:00 [INFO] Successfully inserted 50 users.
2023-10-24 10:00:00 [INFO] Generating 8 categories...
2023-10-24 10:00:00 [INFO] Successfully inserted 8 categories.
2023-10-24 10:00:01 [INFO] Generating approx 250 posts (5 per user)...
2023-10-24 10:00:01 [INFO] Successfully inserted 250 posts.
2023-10-24 10:00:01 [INFO] Generating approx 1000 comments...
2023-10-24 10:00:01 [INFO] Successfully inserted 1000 comments.
2023-10-24 10:00:01 [INFO] Seeding completed successfully!
2023-10-24 10:00:01 [INFO] Database connection closed.
Pass customized environment variables, target a specific YAML configuration, and supply the --clean flag to reset existing tables:
DB_HOST="staging-db.internal" \
DB_NAME="staging_app" \
DB_USER="seeder_bot" \
DB_PASSWORD="securepassword" \
python3 dev_seeder.py --config staging_config.yaml --env staging --clean
Output:
2023-10-24 10:05:00 [INFO] Starting database seeder [Environment: staging]
2023-10-24 10:05:00 [INFO] Successfully loaded configuration from staging_config.yaml
2023-10-24 10:05:00 [INFO] Connected to PostgreSQL database 'staging_app' at staging-db.internal:5432
2023-10-24 10:05:00 [INFO] Validated database schema (created tables if missing).
2023-10-24 10:05:00 [WARNING] Truncating