Migrations#

Quickstart#

Export a configuration from a module the CLI can import. Importing this module only defines the configuration -- migrations run when you invoke the CLI, not at application import time.

from sqlspec.adapters.sqlite import SqliteConfig

database_config = SqliteConfig(
    bind_key="app",
    connection_config={"database": "app.db"},
    migration_config={"script_location": "migrations", "version_table_name": "schema_versions"},
)

Point the CLI at that object with module:attribute and run the workflow:

sqlspec --config database:database_config show-config
sqlspec --config database:database_config init --no-prompt
sqlspec --config database:database_config create-migration -m "create users table" --no-prompt
sqlspec --config database:database_config upgrade --no-prompt
sqlspec --config database:database_config show-current-revision

show-config is the fastest way to confirm the CLI found what you expected before running anything that touches the database.

This example and the full command sequence are exercised by the documentation test suite.

Pointing the CLI at your configuration#

The reference names a module and the attribute holding your configuration. Both separators work, so database:database_config and database.database_config are equivalent:

sqlspec --config database:database_config show-config

The attribute may be a single configuration, a list of configurations, or a factory function returning either. Naming the module alone is not enough -- SQLSpec reports the references the module exports so you can correct the command.

To avoid repeating --config, set an environment variable:

export SQLSPEC_CONFIG=database:database_config
sqlspec show-config

Or record it once in pyproject.toml:

[tool.sqlspec]
config = "database:database_config"

--config wins over SQLSPEC_CONFIG, which wins over pyproject.toml.

To manage several databases at once, separate references with commas. Configurations are deduplicated by bind_key, so give each one a distinct key:

sqlspec --config database:primary_config,database:replica_config upgrade

Modules are imported from the current working directory, so a database.py beside your pyproject.toml is importable without installing your project.

Configuration#

migration_config customizes script locations, the version table, and extension behavior. Unrecognized keys raise ImproperConfigurationError at construction rather than being silently ignored, so a misspelling surfaces immediately.

from sqlspec.adapters.duckdb import DuckDBConfig

config = DuckDBConfig(
    connection_config={"database": "/tmp/analytics.db"},
    migration_config={
        "script_location": "migrations/duckdb",
        "version_table_name": "_schema_versions",
    },
)

Migrations can also be driven in process:

# Apply all pending migrations (head)
config.migrate_up()

# Apply up to a specific revision
config.migrate_up(revision="003")

# Dry-run migration execution
config.migrate_up(dry_run=True)

# Revert the last migration step
config.migrate_down()

# Revert back to a specific revision (or "base" for all)
config.migrate_down(revision="base")

For async configurations, migrate_up() and migrate_down() return awaitables:

from sqlspec.adapters.asyncpg import AsyncpgConfig

config = AsyncpgConfig(
    connection_config={"dsn": "postgresql://localhost/app"},
    migration_config={"script_location": "migrations/postgres"},
)

await config.migrate_up()
await config.migrate_down(revision="-1")

from sqlspec.adapters.sqlite import SqliteConfig

config = SqliteConfig(
    connection_config={"database": ":memory:"}, migration_config={"script_location": "migrations"}
)

# Check custom tracker setup and extension discovery before serving requests.
commands = config.get_migration_commands()

try:
    with config.provide_session() as session:
        value = session.select_value("SELECT 1")
finally:
    config.close_pool()

Running Against an Existing Schema#

Set default_schema when migration SQL should run against a pre-existing schema without qualifying every table in every migration file. SQLSpec validates the schema before creating the tracker table or applying DDL, then configures the session before each migration runs.

Set version_table_schema when the tracker table belongs somewhere other than the objects being migrated. It falls back to default_schema; if neither is set, the tracker table is unqualified and uses the adapter's normal default namespace.

from sqlspec.adapters.asyncpg import AsyncpgConfig

config = AsyncpgConfig(
    connection_config={"dsn": "postgresql://localhost/app"},
    migration_config={
        "script_location": "migrations/postgres",
        "version_table_name": "schema_versions",
        "default_schema": "app_schema",
        "version_table_schema": "admin_schema",
    },
)

Create the target schema before running migrations. The migration role needs the database-specific privileges to create objects there -- for PostgreSQL, usually USAGE and CREATE on the target schema plus permission to create or update the tracker table.

Example with unqualified DDL:

    from sqlspec.adapters.duckdb import DuckDBConfig
    from sqlspec.migrations.commands import SyncMigrationCommands

    migration_dir = tmp_path / "migrations"
    db_path = tmp_path / "app.duckdb"

    config = DuckDBConfig(
        connection_config={"database": str(db_path)},
        migration_config={
            "script_location": str(migration_dir),
            "version_table_name": "schema_versions",
            "default_schema": "app_schema",
            "version_table_schema": "admin_schema",
        },
    )

    try:
        with config.provide_session() as session:
            session.execute("CREATE SCHEMA app_schema")
            session.execute("CREATE SCHEMA admin_schema")

        commands = SyncMigrationCommands(config)
        commands.init(str(migration_dir), package=True)

        (migration_dir / "0001_create_users.py").write_text(
            '''"""Create users."""


def up():
    """Create an unqualified table in app_schema."""
    return ["CREATE TABLE users (id INTEGER PRIMARY KEY, name VARCHAR NOT NULL)"]


def down():
    """Drop the unqualified table from app_schema."""
    return ["DROP TABLE IF EXISTS users"]
'''
        )

        commands.upgrade()

        with config.provide_session() as session:
            users_table = session.select_value(
                """
                SELECT table_name
                FROM information_schema.tables
                WHERE table_schema = ? AND table_name = ?
                """,
                "app_schema",
                "users",
            )
            tracker_table = session.select_value(
                """
                SELECT table_name
                FROM information_schema.tables
                WHERE table_schema = ? AND table_name = ?
                """,
                "admin_schema",
                "schema_versions",
            )

        assert users_table == "users"
        assert tracker_table == "schema_versions"
    finally:
        if config.connection_instance:
            config.close_pool()

-- schema: analytics
-- name: create-reports-table
CREATE TABLE reports (
    id INT PRIMARY KEY,
    title NVARCHAR(255)
);

Extension Migrations#

Extensions are auto-included when a matching entry exists in extension_config. Names resolve against the sqlspec.extensions.<name> namespace by default.

A package distributed separately from SQLSpec points at its own migrations directory with migrations_path:

config = AsyncpgConfig(
    connection_config={"dsn": "postgresql://localhost/app"},
    extension_config={
        "litestar_queues": {
            "migrations_path": "litestar_queues.backends.sqlspec:migrations",
            "table_name": "queue_tasks",
        }
    },
)

Declaring migrations_path auto-includes the extension, so it does not also need to appear in include_extensions. exclude_extensions still opts it back out. The value takes either form:

• A '<dotted.module>:<subdir>' specification resolved against the installed package. Portable across machines, so prefer it in pyproject.toml.

• A filesystem path, absolute or relative to the working directory, matching how script_location resolves.

Packages that register migrations at runtime can call add_extension_migrations instead of declaring the key:

from pathlib import Path

config.add_extension_migrations(
    "litestar_queues",
    Path(__file__).parent / "migrations",
    settings={"table_name": "queue_tasks"},
)

To unregister an extension's migrations at runtime, call remove_extension_migrations:

removed = config.remove_extension_migrations("litestar_queues")

These methods update the extension entry under extension_config and migration_config["include_extensions"]. When an entry changes, cached migration commands are cleared and rebuilt on their next access. The existing SQL loader and its loaded queries are preserved. Mutating extension_config or runner internals directly does not re-run discovery.

SQL files inside a registered extension directory use their filename-local version in named-query directives. For example, migrations/0001_create_queue.sql declares migrate-0001-up and migrate-0001-down. SQLSpec still records that migration as ext_litestar_queues_0001 when the registered extension name is litestar_queues.

An extension migration stored in the application's main migration directory instead carries the prefix in its filename, such as ext_litestar_queues_0001_create_queue.sql, and therefore declares migrate-ext_litestar_queues_0001-up and migrate-ext_litestar_queues_0001-down.

Migration File Templates#

create-migration renders new files from a built-in template. Three migration_config keys adjust what it writes:

Overrides replace individual fragments; anything omitted keeps its default:

config = DuckDBConfig(
    connection_config={"database": "/tmp/analytics.db"},
    migration_config={
        "title": "Acme Migration",
        "default_format": "py",
        "templates": {
            "sql": {
                "header": "-- {title} [{adapter}]",
                "metadata": ["-- Version: {version}", "-- Owner: {author}"],
            }
        },
    },
)

Every fragment is rendered with str.format, so these placeholders are available: title, version, message, description, created_at, author, adapter, project_slug, and slug (the filename-safe form of the message). An unknown placeholder raises TemplateValidationError when the file is generated.

The SQL template accepts header, metadata, body, and description_key; the Python template accepts docstring, imports, body, and description_key. description_key names the label the description is read back from, and takes a string or a list of strings.

See MigrationTemplates for the full override shape.

Return Values in Python Migrations#

up() and down() return either a string or an iterable of strings. Each statement is executed in order, and blank or whitespace-only statements are skipped. Returning anything else raises MigrationLoadError.

Returning an empty list executes nothing, but the migration is still recorded in the tracking table and is not reported as pending again. This is the supported way to write a conditional migration that inspects the database and finds nothing to do:

def up(context: object | None = None) -> list[str]:
    """Add the audit column only where it is missing."""
    if _has_audit_column(context):
        return []
    return ["ALTER TABLE orders ADD COLUMN audited_at TIMESTAMP"]

Programmatic Migration Runner#

In addition to config.migrate_up(), SQLSpec provides lower-level migration command interfaces via create_migration_commands(config) or config.get_migration_commands():

from sqlspec.migrations.commands import create_migration_commands

commands = create_migration_commands(config)

# Sync configs: direct execution
commands.init("migrations", package=True)
commands.revision("add products table", file_type="sql")
commands.upgrade(revision="head")
current_rev = commands.current(verbose=True)
commands.stamp("0003")

For async configurations, ``create_migration_commands`` returns an ``AsyncMigrationCommands`` instance where operations like ``init()``, ``upgrade()``, ``downgrade()``, ``revision()``, and ``current()`` are awaitable. In addition, configurations provide ``config.get_current_migration(verbose=False)`` for direct revision checks.

Squashing Migrations#

Over time, long migration histories can slow down fresh database provisioning and test initialization. SQLSpec can squash a contiguous range of sequential migrations into a single consolidated migration file:

# Squash revisions 0001 through 0007 into a single migration
sqlspec squash 1:7 -m "squash initial schema" --dry-run
sqlspec squash 1:7 -m "squash initial schema" --yes

In Python:

commands.squash(
    start_version="0001",
    end_version="0007",
    description="squash initial schema",
    output_format="sql",  # or "py"
    dry_run=False,
)

The squashed migration combines the SQL statements, updates tracker records, and safely removes the intermediate files. Use --allow-gaps if migrations contain gaps in their version numbers.

Fixing Timestamp Migrations#

If your project began with timestamp-based migration names (e.g., 20240101120000_create_users.sql), use the fix command to convert them to deterministic sequential format (0001_create_users.sql):

sqlspec fix --dry-run
sqlspec fix --yes

This updates both the filenames on disk and the corresponding applied records in the database tracking table.

Additive Schema Management#

For microservices or applications requiring additive schema verification at startup without running full migration files, SQLSpec includes ensure_schema_sync and ensure_schema_async:

from sqlspec.migrations.schema import SchemaTarget, ensure_schema_sync
from sqlspec.builder import sql

# Define target table using SQLSpec builder
users_target = SchemaTarget(
    table_name="users",
    create_table=(
        sql.create_table("users")
        .column("id", "INTEGER", primary_key=True)
        .column("email", "VARCHAR(255)", not_null=True, unique=True)
        .column("created_at", "TIMESTAMP", default="CURRENT_TIMESTAMP")
    ),
)

with config.provide_session() as driver:
    result = ensure_schema_sync(
        driver,
        [users_target],
        manage_schema=True,  # Discovers existing schema and creates missing tables/columns
        create_schema=True,
    )
    print(f"Created tables: {result.created_tables}, Added columns: {result.added_columns}")

Output and Logging#

Control output with migration_config keys or their CLI equivalents:

Related Guides#

• Command Line Interface for full CLI usage.

• Migrations for API details.