Cookbook
Recipes for common patterns that go beyond the basics.
Cyclic foreign keys
Some schemas have mutually referencing tables — for example, a Company that has a ceo_id FK to Person, and a Person that has an employer_id FK to Company. Standard insertion order cannot satisfy both constraints simultaneously.
Solution: defer FK constraints (PostgreSQL), then break the cycle with a two-step insert.
from seedling import Seeder, deferred_constraints, DEV_AND_TEST
from sqlalchemy import update
from sqlalchemy.ext.asyncio import AsyncSession
from myapp.models import Company, Person
class OrgChartSeeder(Seeder):
environments = DEV_AND_TEST
models = [Company, Person]
async def run(self, session: AsyncSession) -> None:
async with deferred_constraints(session):
# 1. Insert companies with no CEO yet
acme = Company(name="Acme Corp", ceo_id=None)
session.add(acme)
await session.flush()
# 2. Insert persons referencing the company
alice = Person(name="Alice", employer_id=acme.id)
session.add(alice)
await session.flush()
# 3. Update the company's CEO now that Alice has an id
acme.ceo_id = alice.id
await session.commit()deferred_constraints() is a no-op on SQLite and MySQL/MariaDB — for those dialects, break the cycle by inserting with NULL and updating in a second statement (same pattern, no context manager needed).
Polymorphic models
Single-table inheritance (STI)
SQLAlchemy STI uses a type discriminator column to select the concrete class. Create one factory per concrete subclass:
from sqlalchemy.orm import DeclarativeBase, mapped_column
from sqlalchemy import String
from seedling import Factory, Faker
class Base(DeclarativeBase):
pass
class Employee(Base):
__tablename__ = "employee"
id: int = mapped_column(primary_key=True)
name: str = mapped_column(String)
type: str = mapped_column(String)
__mapper_args__ = {"polymorphic_on": "type", "polymorphic_identity": "employee"}
class Manager(Employee):
reports_count: int = mapped_column(default=0)
__mapper_args__ = {"polymorphic_identity": "manager"}
class Engineer(Employee):
stack: str = mapped_column(String, nullable=True)
__mapper_args__ = {"polymorphic_identity": "engineer"}
class ManagerFactory(Factory[Manager]):
model = Manager
name = Faker("name")
class EngineerFactory(Factory[Engineer]):
model = Engineer
name = Faker("name")
stack = "Python"Each factory targets the concrete subclass. SQLAlchemy fills in type automatically from __mapper_args__["polymorphic_identity"].
Joined-table inheritance (JTI)
JTI works the same way — declare a factory for each concrete subclass. The base-table columns are inherited through the normal SQLAlchemy mapper chain and you don't need to do anything special in the factory.
JSONB and arrays (PostgreSQL)
PostgreSQL JSONB and ARRAY columns are not covered by AutoFactory smart defaults. Declare them explicitly:
from seedling import Factory, Faker, LazyAttribute
from myapp.models import Article
class ArticleFactory(Factory[Article]):
model = Article
title = Faker("sentence", nb_words=6)
# JSONB column
metadata = LazyAttribute(
lambda f: {
"tags": ["python", "async"],
"source": "import",
"version": 1,
}
)
# ARRAY column
keywords = LazyAttribute(
lambda f: [
Faker("word").generate({}),
Faker("word").generate({}),
]
)For randomised JSONB content, use faker inside LazyAttribute:
from seedling import faker
metadata = LazyAttribute(
lambda f: {
"score": faker.random_int(0, 100),
"tags": faker.words(3),
"created": faker.iso8601(),
}
)Time-series data
Seed rows with monotonically increasing or decreasing timestamps using Sequence and the datetime module:
from datetime import UTC, datetime, timedelta
from seedling import Factory, Sequence
from myapp.models import Event
_BASE_TIME = datetime(2026, 1, 1, tzinfo=UTC)
class EventFactory(Factory[Event]):
model = Event
name = Sequence(lambda n: f"event-{n:04d}")
# Each event is 1 hour after the previous one
occurred_at = Sequence(lambda n: _BASE_TIME + timedelta(hours=n))For the last N days of data, compute from the current time:
from datetime import UTC, datetime, timedelta
from seedling import Factory, Sequence
class MetricFactory(Factory[Metric]):
model = Metric
value = Sequence(lambda n: float(n % 100))
recorded_at = Sequence(
lambda n: datetime.now(UTC) - timedelta(days=30) + timedelta(hours=n)
)To seed a full 30-day window in a seeder:
class MetricsSeeder(Seeder):
environments = DEV_AND_TEST
models = [Metric]
async def run(self, session: AsyncSession) -> None:
await MetricFactory.create_batch(session, 30 * 24) # one per hourLarge tables
Use bulk=True for 10 000+ rows
create_batch(session, n, bulk=True) uses a single INSERT ... RETURNING statement instead of N per-row round trips. It is significantly faster for large batches.
await UserFactory.create_batch(session, 100_000, bulk=True)Limitations: @post_generation hooks and RelatedFactory / RelatedFactoryList do not fire. SubFactory and FK auto-resolve fields are omitted — pass them as explicit overrides:
await PostFactory.create_batch(session, 50_000, bulk=True, author_id=fixed_user_id)Disable state tracking for pure performance runs
State tracking opens extra sessions per seeder to write state rows. Disable it when you only care about speed:
runner = SeederRunner(session_factory, env="staging", state_tracking=False)Or per-project:
[tool.seedling]
state_tracking = falseChunked seeding for very large datasets
If you need to seed tens of millions of rows, chunk the work to avoid memory pressure:
class BigTableSeeder(Seeder):
environments = DEV_AND_TEST
models = [Event]
CHUNK_SIZE = 10_000
TOTAL = 1_000_000
async def run(self, session: AsyncSession) -> None:
for _ in range(self.TOTAL // self.CHUNK_SIZE):
await EventFactory.create_batch(session, self.CHUNK_SIZE, bulk=True)Each create_batch call issues one INSERT statement; the loop keeps peak memory proportional to CHUNK_SIZE, not TOTAL.
Parallel seeders for independent tables
If you have multiple large tables with no FK relationship, register their seeders without depends_on — the runner will execute them in parallel:
class UserSeeder(Seeder):
environments = DEV_AND_TEST
models = [User]
async def run(self, session):
await UserFactory.create_batch(session, 100_000, bulk=True)
class ProductSeeder(Seeder):
environments = DEV_AND_TEST
models = [Product]
async def run(self, session):
await ProductFactory.create_batch(session, 500_000, bulk=True)
# No depends_on — these run in parallel
runner.register(UserSeeder, ProductSeeder)Alembic post-upgrade hook
Run seeders automatically after Alembic migrations by hooking into env.py:
from alembic import context
def run_migrations_online() -> None:
# ... standard async alembic setup ...
with connectable.connect() as connection:
context.configure(connection=connection, target_metadata=target_metadata)
with context.begin_transaction():
context.run_migrations()
# After migrations complete, run seeders in the test environment
if context.get_x_argument(as_dictionary=True).get("seed"):
import asyncio
from myapp.seeders import create_runner
asyncio.run(create_runner("test").run())Trigger it via:
alembic -x seed=true upgrade headTransactional test fixtures
Use seedling_transactional_session to get a session that auto-rolls back after each test, keeping the database clean without truncation:
import pytest
from sqlalchemy.ext.asyncio import async_sessionmaker, create_async_engine
@pytest.fixture(scope="session")
def seedling_session_factory():
engine = create_async_engine("sqlite+aiosqlite:///:memory:")
return async_sessionmaker(engine, expire_on_commit=False)
# test_orders.py
from seedling.pytest_plugin import seed
@seed(UserSeeder, OrderSeeder)
async def test_order_total(seedling_transactional_session):
session = seedling_transactional_session
# Both seeders ran before this test body.
# Any changes made in this test are rolled back automatically.
orders = await session.execute(select(Order))
...