Database Seeding and Fixture Loading¶

Advanced Alchemy provides utilities for seeding your database with initial data through JSON fixtures. This documentation will show you how to create and load fixtures in both synchronous and asynchronous applications.

Creating Fixtures¶

Fixtures in Advanced Alchemy are simple JSON files that contain the data you want to seed. Each fixture file should:

Contain a JSON object or a JSON array of objects, where each object represents a row in your database table
Include all required fields for your model

Example Fixture:

fixtures/products.json¶

[
    {
        "name": "Laptop",
        "description": "High-performance laptop with 16GB RAM and 1TB SSD",
        "price": 999.99,
        "in_stock": true
    },
    {
        "name": "Smartphone",
        "description": "Latest smartphone model with 5G and advanced camera",
        "price": 699.99,
        "in_stock": true
    },
    {
        "name": "Headphones",
        "description": "Noise-cancelling wireless headphones with 30-hour battery life",
        "price": 199.99,
        "in_stock": true
    },
    {
        "name": "Smartwatch",
        "description": "Fitness tracker with heart rate monitor and GPS",
        "price": 149.99,
        "in_stock": false
    },
    {
        "name": "Tablet",
        "description": "10-inch tablet with high-resolution display",
        "price": 349.99,
        "in_stock": true
    }
]

Loading Fixtures¶

Advanced Alchemy provides both synchronous and asynchronous functions for loading fixtures:

Synchronous Loading¶

from pathlib import Path

from sqlalchemy import String, create_engine
from sqlalchemy.orm import Mapped, mapped_column

from advanced_alchemy.base import UUIDBase
from advanced_alchemy.config import SQLAlchemySyncConfig, SyncSessionConfig
from advanced_alchemy.repository import SQLAlchemySyncRepository
from advanced_alchemy.utils.fixtures import open_fixture

# Database connection string
DATABASE_URL = "sqlite:///db.sqlite3"

config = SQLAlchemySyncConfig(
    engine_instance=create_engine(DATABASE_URL),
    session_config=SyncSessionConfig(expire_on_commit=False)
)

class Product(UUIDBase):
    """Product model."""
    __tablename__ = "products"

    name: Mapped[str] = mapped_column(String(length=100))
    description: Mapped[str | None] = mapped_column(String(length=500))
    price: Mapped[float]
    in_stock: Mapped[bool] = mapped_column(default=True)


# Repository
class ProductRepository(SQLAlchemySyncRepository[Product]):
    """Product repository."""
    model_type = Product


# Set up fixtures path
fixtures_path = Path(__file__).parent / "fixtures"


def initialize_database():
    """Initialize the database and create tables."""
    print("Creating database tables...")
    with config.get_engine().begin() as conn:
        UUIDBase.metadata.create_all(conn)
    print("Tables created successfully")


def seed_database():
    """Seed the database with fixture data."""
    print("Seeding database...")

    # Create a session
    with config.get_session() as session:
        # Create repository for product model
        product_repo = ProductRepository(session=session)

        # Load and add product data
        try:
            print(f"Attempting to load fixtures from {fixtures_path}/product.json")
            product_data = open_fixture(fixtures_path, "product")
            print(f"Loaded {len(product_data)} products from fixture")
            product_repo.add_many([Product(**item) for item in product_data])
            session.commit()
        except FileNotFoundError:
            print(f"Could not find fixture file at {fixtures_path}/product.json")


if __name__ == "__main__":
    # Initialize the database
    initialize_database()

    # Seed the database
    seed_database()

Asynchronous Loading¶

Python 3.9+

import asyncio
from pathlib import Path
from typing import Optional

from sqlalchemy import String
from sqlalchemy.ext.asyncio import create_async_engine
from sqlalchemy.orm import Mapped, mapped_column

from advanced_alchemy.base import UUIDBase
from advanced_alchemy.config import AsyncSessionConfig, SQLAlchemyAsyncConfig
from advanced_alchemy.repository import SQLAlchemyAsyncRepository
from advanced_alchemy.utils.fixtures import open_fixture_async

# Database connection string
DATABASE_URL = "sqlite+aiosqlite:///db.sqlite3"

config = SQLAlchemyAsyncConfig(
    engine_instance=create_async_engine(DATABASE_URL),
    session_config=AsyncSessionConfig(expire_on_commit=False)
)

class Product(UUIDBase):
    """Product model."""
    __tablename__ = "products"

    name: Mapped[str] = mapped_column(String(length=100))
    description: Mapped[Optional[str]] = mapped_column(String(length=500))
    price: Mapped[float]
    in_stock: Mapped[bool] = mapped_column(default=True)


# Repository
class ProductRepository(SQLAlchemyAsyncRepository[Product]):
    """Product repository."""
    model_type = Product


# Set up fixtures path
fixtures_path = Path(__file__).parent / "fixtures"


async def initialize_database():
    """Initialize the database and create tables."""
    print("Creating database tables...")
    async with config.get_engine().begin() as conn:
        await conn.run_sync(UUIDBase.metadata.create_all)
    print("Tables created successfully")


async def seed_database():
    """Seed the database with fixture data."""
    print("Seeding database...")

    # Create a session
    async with config.get_session() as session:
        # Create repository for product model
        product_repo = ProductRepository(session=session)

        # Load and add product data
        try:
            print(f"Attempting to load fixtures from {fixtures_path}/product.json")
            product_data = await open_fixture_async(fixtures_path, "product")
            print(f"Loaded {len(product_data)} products from fixture")
            await product_repo.add_many([Product(**item) for item in product_data])
            await session.commit()
        except FileNotFoundError:
            print(f"Could not find fixture file at {fixtures_path}/product.json")



async def main():
    """Main async function to run the example."""
    # Initialize the database
    await initialize_database()

    # Seed the database
    await seed_database()



if __name__ == "__main__":
    # Run the async main function
    asyncio.run(main())

Python 3.10+

import asyncio
from pathlib import Path

from sqlalchemy import String
from sqlalchemy.ext.asyncio import create_async_engine
from sqlalchemy.orm import Mapped, mapped_column

from advanced_alchemy.base import UUIDBase
from advanced_alchemy.config import AsyncSessionConfig, SQLAlchemyAsyncConfig
from advanced_alchemy.repository import SQLAlchemyAsyncRepository
from advanced_alchemy.utils.fixtures import open_fixture_async

# Database connection string
DATABASE_URL = "sqlite+aiosqlite:///db.sqlite3"

config = SQLAlchemyAsyncConfig(
    engine_instance=create_async_engine(DATABASE_URL),
    session_config=AsyncSessionConfig(expire_on_commit=False)
)

class Product(UUIDBase):
    """Product model."""
    __tablename__ = "products"

    name: Mapped[str] = mapped_column(String(length=100))
    description: Mapped[str | None] = mapped_column(String(length=500))
    price: Mapped[float]
    in_stock: Mapped[bool] = mapped_column(default=True)


# Repository
class ProductRepository(SQLAlchemyAsyncRepository[Product]):
    """Product repository."""
    model_type = Product


# Set up fixtures path
fixtures_path = Path(__file__).parent / "fixtures"


async def initialize_database():
    """Initialize the database and create tables."""
    print("Creating database tables...")
    async with config.get_engine().begin() as conn:
        await conn.run_sync(UUIDBase.metadata.create_all)
    print("Tables created successfully")


async def seed_database():
    """Seed the database with fixture data."""
    print("Seeding database...")

    # Create a session
    async with config.get_session() as session:
        # Create repository for product model
        product_repo = ProductRepository(session=session)

        # Load and add product data
        try:
            print(f"Attempting to load fixtures from {fixtures_path}/product.json")
            product_data = await open_fixture_async(fixtures_path, "product")
            print(f"Loaded {len(product_data)} products from fixture")
            await product_repo.add_many([Product(**item) for item in product_data])
            await session.commit()
        except FileNotFoundError:
            print(f"Could not find fixture file at {fixtures_path}/product.json")



async def main():
    """Main async function to run the example."""
    # Initialize the database
    await initialize_database()

    # Seed the database
    await seed_database()



if __name__ == "__main__":
    # Run the async main function
    asyncio.run(main())

Integration with Web Frameworks¶

Litestar¶

Python 3.9+

from pathlib import Path
from typing import Optional

import uvicorn
from litestar import Litestar
from sqlalchemy import String
from sqlalchemy.orm import Mapped, mapped_column

from advanced_alchemy.base import UUIDBase
from advanced_alchemy.extensions.litestar import (
    AsyncSessionConfig,
    SQLAlchemyAsyncConfig,
    SQLAlchemyPlugin,
)
from advanced_alchemy.repository import SQLAlchemyAsyncRepository
from advanced_alchemy.utils.fixtures import open_fixture_async

# Database connection string
DATABASE_URL = "sqlite+aiosqlite:///db.sqlite3"

# Set up fixtures path
fixtures_path = Path(__file__).parent / "fixtures"

session_config = AsyncSessionConfig(expire_on_commit=False)
sqlalchemy_config = SQLAlchemyAsyncConfig(
    connection_string=DATABASE_URL,
    before_send_handler="autocommit",
    session_config=session_config,
    create_all=True,
)
alchemy = SQLAlchemyPlugin(config=sqlalchemy_config)


class Product(UUIDBase):
    """Product model."""
    __tablename__ = "products"

    name: Mapped[str] = mapped_column(String(length=100))
    description: Mapped[Optional[str]] = mapped_column(String(length=500))
    price: Mapped[float]
    in_stock: Mapped[bool] = mapped_column(default=True)


# Repository
class ProductRepository(SQLAlchemyAsyncRepository[Product]):
    """Product repository."""
    model_type = Product


# Startup function to seed the database
async def on_startup() -> None:
    """Seed the database during application startup."""
    print("Running startup routine...")

    # Create a session and seed data
    async with sqlalchemy_config.get_session() as session:
        # Create repository for product model
        product_repo = ProductRepository(session=session)
        # Load and add product data
        try:
            print(f"Attempting to load fixtures from {fixtures_path}/product.json")
            product_data = await open_fixture_async(fixtures_path, "product")
            print(f"Loaded {len(product_data)} products from fixture")
            await product_repo.add_many([Product(**item) for item in product_data])
            await session.commit()
        except FileNotFoundError:
            print(f"Could not find fixture file at {fixtures_path}/product.json")

        # Verify data was added
        products = await product_repo.list()
        print(f"Database seeded with {len(products)} products")


# Create the Litestar application
app = Litestar(
    on_startup=[on_startup],
    plugins=[alchemy],
)

if __name__ == "__main__":
    uvicorn.run(app, host="0.0.0.0", port=8000)

Python 3.10+

from pathlib import Path

import uvicorn
from litestar import Litestar
from sqlalchemy import String
from sqlalchemy.orm import Mapped, mapped_column

from advanced_alchemy.base import UUIDBase
from advanced_alchemy.extensions.litestar import (
    AsyncSessionConfig,
    SQLAlchemyAsyncConfig,
    SQLAlchemyPlugin,
)
from advanced_alchemy.repository import SQLAlchemyAsyncRepository
from advanced_alchemy.utils.fixtures import open_fixture_async

# Database connection string
DATABASE_URL = "sqlite+aiosqlite:///db.sqlite3"

# Set up fixtures path
fixtures_path = Path(__file__).parent / "fixtures"

session_config = AsyncSessionConfig(expire_on_commit=False)
sqlalchemy_config = SQLAlchemyAsyncConfig(
    connection_string=DATABASE_URL,
    before_send_handler="autocommit",
    session_config=session_config,
    create_all=True,
)
alchemy = SQLAlchemyPlugin(config=sqlalchemy_config)


class Product(UUIDBase):
    """Product model."""
    __tablename__ = "products"

    name: Mapped[str] = mapped_column(String(length=100))
    description: Mapped[str | None] = mapped_column(String(length=500))
    price: Mapped[float]
    in_stock: Mapped[bool] = mapped_column(default=True)


# Repository
class ProductRepository(SQLAlchemyAsyncRepository[Product]):
    """Product repository."""
    model_type = Product


# Startup function to seed the database
async def on_startup() -> None:
    """Seed the database during application startup."""
    print("Running startup routine...")

    # Create a session and seed data
    async with sqlalchemy_config.get_session() as session:
        # Create repository for product model
        product_repo = ProductRepository(session=session)
        # Load and add product data
        try:
            print(f"Attempting to load fixtures from {fixtures_path}/product.json")
            product_data = await open_fixture_async(fixtures_path, "product")
            print(f"Loaded {len(product_data)} products from fixture")
            await product_repo.add_many([Product(**item) for item in product_data])
            await session.commit()
        except FileNotFoundError:
            print(f"Could not find fixture file at {fixtures_path}/product.json")

        # Verify data was added
        products = await product_repo.list()
        print(f"Database seeded with {len(products)} products")


# Create the Litestar application
app = Litestar(
    on_startup=[on_startup],
    plugins=[alchemy],
)

if __name__ == "__main__":
    uvicorn.run(app, host="0.0.0.0", port=8000)

FastAPI¶

Python 3.9+

from contextlib import asynccontextmanager
from pathlib import Path
from typing import Optional

import uvicorn
from fastapi import FastAPI
from sqlalchemy import String
from sqlalchemy.orm import Mapped, mapped_column

from advanced_alchemy.base import UUIDBase
from advanced_alchemy.extensions.fastapi import (
    AdvancedAlchemy,
    AsyncSessionConfig,
    SQLAlchemyAsyncConfig,
)
from advanced_alchemy.repository import SQLAlchemyAsyncRepository
from advanced_alchemy.utils.fixtures import open_fixture_async

# Database connection string
DATABASE_URL = "sqlite+aiosqlite:///db.sqlite3"

# Set up fixtures path
fixtures_path = Path(__file__).parent / "fixtures"


class Product(UUIDBase):
    """Product model."""
    __tablename__ = "products"

    name: Mapped[str] = mapped_column(String(length=100))
    description: Mapped[Optional[str]] = mapped_column(String(length=500))
    price: Mapped[float]
    in_stock: Mapped[bool] = mapped_column(default=True)


# Repository
class ProductRepository(SQLAlchemyAsyncRepository[Product]):
    """Product repository."""
    model_type = Product


# Lifespan context manager
@asynccontextmanager
async def lifespan(app: FastAPI):
    """Handle startup and shutdown events."""
    # Startup: Initialize database and seed data
    print("Running startup routine...")

    # Create a session and seed data
    async with sqlalchemy_config.get_session() as session:
        # Create repository for product model
        product_repo = ProductRepository(session=session)
        # Load and add product data
        try:
            print(f"Attempting to load fixtures from {fixtures_path}/product.json")
            product_data = await open_fixture_async(fixtures_path, "product")
            print(f"Loaded {len(product_data)} products from fixture")
            await product_repo.add_many([Product(**item) for item in product_data])
            await session.commit()
        except FileNotFoundError:
            print(f"Could not find fixture file at {fixtures_path}/product.json")

        # Verify data was added
        products = await product_repo.list()
        print(f"Database seeded with {len(products)} products")

    # Yield control back to FastAPI
    yield

    # Shutdown: Clean up resources if needed
    # This section runs when the application is shutting down
    print("Shutting down...")


session_config = AsyncSessionConfig(expire_on_commit=False)
sqlalchemy_config = SQLAlchemyAsyncConfig(
    connection_string=DATABASE_URL,
    commit_mode="autocommit",
    session_config=session_config,
    create_all=True,
)

# Create the FastAPI application with lifespan
app = FastAPI(lifespan=lifespan)

alchemy = AdvancedAlchemy(config=sqlalchemy_config, app=app)

if __name__ == "__main__":
    uvicorn.run(app, host="0.0.0.0", port=8000)

Python 3.10+

from contextlib import asynccontextmanager
from pathlib import Path

import uvicorn
from fastapi import FastAPI
from sqlalchemy import String
from sqlalchemy.orm import Mapped, mapped_column

from advanced_alchemy.base import UUIDBase
from advanced_alchemy.extensions.fastapi import (
    AdvancedAlchemy,
    AsyncSessionConfig,
    SQLAlchemyAsyncConfig,
)
from advanced_alchemy.repository import SQLAlchemyAsyncRepository
from advanced_alchemy.utils.fixtures import open_fixture_async

# Database connection string
DATABASE_URL = "sqlite+aiosqlite:///db.sqlite3"

# Set up fixtures path
fixtures_path = Path(__file__).parent / "fixtures"


class Product(UUIDBase):
    """Product model."""
    __tablename__ = "products"

    name: Mapped[str] = mapped_column(String(length=100))
    description: Mapped[str | None] = mapped_column(String(length=500))
    price: Mapped[float]
    in_stock: Mapped[bool] = mapped_column(default=True)


# Repository
class ProductRepository(SQLAlchemyAsyncRepository[Product]):
    """Product repository."""
    model_type = Product


# Lifespan context manager
@asynccontextmanager
async def lifespan(app: FastAPI):
    """Handle startup and shutdown events."""
    # Startup: Initialize database and seed data
    print("Running startup routine...")

    # Create a session and seed data
    async with sqlalchemy_config.get_session() as session:
        # Create repository for product model
        product_repo = ProductRepository(session=session)
        # Load and add product data
        try:
            print(f"Attempting to load fixtures from {fixtures_path}/product.json")
            product_data = await open_fixture_async(fixtures_path, "product")
            print(f"Loaded {len(product_data)} products from fixture")
            await product_repo.add_many([Product(**item) for item in product_data])
            await session.commit()
        except FileNotFoundError:
            print(f"Could not find fixture file at {fixtures_path}/product.json")

        # Verify data was added
        products = await product_repo.list()
        print(f"Database seeded with {len(products)} products")

    # Yield control back to FastAPI
    yield

    # Shutdown: Clean up resources if needed
    # This section runs when the application is shutting down
    print("Shutting down...")


session_config = AsyncSessionConfig(expire_on_commit=False)
sqlalchemy_config = SQLAlchemyAsyncConfig(
    connection_string=DATABASE_URL,
    commit_mode="autocommit",
    session_config=session_config,
    create_all=True,
)

# Create the FastAPI application with lifespan
app = FastAPI(lifespan=lifespan)

alchemy = AdvancedAlchemy(config=sqlalchemy_config, app=app)

if __name__ == "__main__":
    uvicorn.run(app, host="0.0.0.0", port=8000)

Flask¶

Python 3.9+

from pathlib import Path
from typing import Optional

from flask import Flask
from sqlalchemy import String
from sqlalchemy.orm import Mapped, mapped_column

from advanced_alchemy.base import UUIDBase
from advanced_alchemy.extensions.flask import (
    AdvancedAlchemy,
    SQLAlchemySyncConfig,
    SyncSessionConfig,
)
from advanced_alchemy.repository import SQLAlchemySyncRepository
from advanced_alchemy.utils.fixtures import open_fixture

# Database connection string
DATABASE_URL = "sqlite:///db.sqlite3"

# Set up fixtures path
fixtures_path = Path(__file__).parent / "fixtures"

class Product(UUIDBase):
    """Product model."""
    __tablename__ = "products"

    name: Mapped[str] = mapped_column(String(length=100))
    description: Mapped[Optional[str]] = mapped_column(String(length=500))
    price: Mapped[float]
    in_stock: Mapped[bool] = mapped_column(default=True)


# Repository
class ProductRepository(SQLAlchemySyncRepository[Product]):
    """Product repository."""
    model_type = Product


app = Flask(__name__)

sqlalchemy_config = SQLAlchemySyncConfig(
    connection_string=DATABASE_URL,
    commit_mode="autocommit",
    session_config=SyncSessionConfig(
        expire_on_commit=False,
    ),
    create_all=True
)

db = AdvancedAlchemy(config=sqlalchemy_config)
db.init_app(app)

with app.app_context():  # noqa: SIM117
    # Seed data
    with db.get_session() as session:
        product_repo = ProductRepository(session=session)
        # Load and add product data
        try:
            print(f"Attempting to load fixtures from {fixtures_path}/product.json")
            product_data = open_fixture(fixtures_path, "product")
            print(f"Loaded {len(product_data)} products from fixture")
            product_repo.add_many([Product(**item) for item in product_data])
            session.commit()
        except FileNotFoundError:
            print(f"Could not find fixture file at {fixtures_path}/product.json")

        # Verify data was added
        products = product_repo.list()
        print(f"Database seeded with {len(products)} products")

if __name__ == "__main__":
    app.run(host="0.0.0.0", port=5000)

Python 3.10+

from pathlib import Path

from flask import Flask
from sqlalchemy import String
from sqlalchemy.orm import Mapped, mapped_column

from advanced_alchemy.base import UUIDBase
from advanced_alchemy.extensions.flask import (
    AdvancedAlchemy,
    SQLAlchemySyncConfig,
    SyncSessionConfig,
)
from advanced_alchemy.repository import SQLAlchemySyncRepository
from advanced_alchemy.utils.fixtures import open_fixture

# Database connection string
DATABASE_URL = "sqlite:///db.sqlite3"

# Set up fixtures path
fixtures_path = Path(__file__).parent / "fixtures"

class Product(UUIDBase):
    """Product model."""
    __tablename__ = "products"

    name: Mapped[str] = mapped_column(String(length=100))
    description: Mapped[str | None] = mapped_column(String(length=500))
    price: Mapped[float]
    in_stock: Mapped[bool] = mapped_column(default=True)


# Repository
class ProductRepository(SQLAlchemySyncRepository[Product]):
    """Product repository."""
    model_type = Product


app = Flask(__name__)

sqlalchemy_config = SQLAlchemySyncConfig(
    connection_string=DATABASE_URL,
    commit_mode="autocommit",
    session_config=SyncSessionConfig(
        expire_on_commit=False,
    ),
    create_all=True
)

db = AdvancedAlchemy(config=sqlalchemy_config)
db.init_app(app)

with app.app_context():  # noqa: SIM117
    # Seed data
    with db.get_session() as session:
        product_repo = ProductRepository(session=session)
        # Load and add product data
        try:
            print(f"Attempting to load fixtures from {fixtures_path}/product.json")
            product_data = open_fixture(fixtures_path, "product")
            print(f"Loaded {len(product_data)} products from fixture")
            product_repo.add_many([Product(**item) for item in product_data])
            session.commit()
        except FileNotFoundError:
            print(f"Could not find fixture file at {fixtures_path}/product.json")

        # Verify data was added
        products = product_repo.list()
        print(f"Database seeded with {len(products)} products")

if __name__ == "__main__":
    app.run(host="0.0.0.0", port=5000)

Best Practices¶

Directory Structure: Keep your fixtures in a dedicated directory (e.g., fixtures/).
Naming Convention: Name your fixture files after the corresponding table names.
Idempotent Seeding: Always check if data exists before seeding to avoid duplicates or update records.
Dependencies: Seed tables in the correct order to respect foreign key constraints.
Data Validation: Ensure your fixture data meets your model’s constraints.
Environment Separation: Consider having different fixtures for development, testing, and production.
Version Control: Keep your fixtures under version control with your application code.

Tips for Efficient Seeding¶

Use add_many (async) / add_many (sync) instead of adding objects one by one for better performance.
Use upsert_many (async) / upsert_many (sync) to update your data if you are updating prices for example.
You can use the database seeding from your cli, app startup or any route.
For large datasets, consider chunking the data into smaller batches.
When dealing with relationships, seed parent records before child records.
Consider using factory libraries like Polyfactory for generating test data.