Migrations¶

Native migration system for SQLSpec that leverages the SQL file loader and driver system for database versioning. Supports SQL and Python migration files, squashing, and validation.

Commands¶

class sqlspec.migrations.SyncMigrationCommands[source]¶

Bases: BaseMigrationCommands[SyncConfigT, Any]

Synchronous migration commands.

__init__(config)[source]¶

Initialize migration commands.

Parameters:: config¶ (TypeVar(SyncConfigT, bound= SyncDatabaseConfig[Any, Any, Any] | NoPoolSyncConfig[Any, Any])) -- The SQLSpec configuration.

init(directory, package=True)[source]¶

Initialize migration directory structure.

Parameters:

directory¶ (str) -- Directory to initialize migrations in.
package¶ (bool) -- Whether to create __init__.py file.

Return type:

None

current(verbose=False)[source]¶

Show current migration version.

Parameters:: verbose¶ (bool) -- Whether to show detailed migration history.
Return type:: str | None
Returns:: The current migration version or None if no migrations applied.

upgrade(revision='head', allow_missing=False, auto_sync=True, dry_run=False, *, use_logger=False, echo=None, summary_only=None)[source]¶

Upgrade to a target revision.

Validates migration order and warns if out-of-order migrations are detected. Out-of-order migrations can occur when branches merge in different orders across environments.

Parameters:

revision¶ (str) -- Target revision or "head" for latest.
allow_missing¶ (bool) -- If True, allow out-of-order migrations even in strict mode. Defaults to False.
auto_sync¶ (bool) -- If True, automatically reconcile renamed migrations in database. Defaults to True. Can be disabled via --no-auto-sync flag.
dry_run¶ (bool) -- If True, show what would be done without making changes.
use_logger¶ (bool) -- If True, output to logger instead of Rich console. Defaults to False. Can be set via MigrationConfig for persistent default.
echo¶ (bool | None) -- Echo output to the console. Defaults to True when unset.
summary_only¶ (bool | None) -- Emit a single summary log entry when logger output is enabled.

Return type:

None

downgrade(revision='-1', *, dry_run=False, use_logger=False, echo=None, summary_only=None)[source]¶

Downgrade to a target revision.

Parameters:

revision¶ (str) -- Target revision or "-1" for one step back.
dry_run¶ (bool) -- If True, show what would be done without making changes.
use_logger¶ (bool) -- If True, output to logger instead of Rich console. Defaults to False. Can be set via MigrationConfig for persistent default.
echo¶ (bool | None) -- Echo output to the console. Defaults to True when unset.
summary_only¶ (bool | None) -- Emit a single summary log entry when logger output is enabled.

Return type:

None

stamp(revision)[source]¶

Mark database as being at a specific revision without running migrations.

Parameters:: revision¶ (str) -- The revision to stamp.
Return type:: None

revision(message, file_type=None)[source]¶

Create a new migration file with timestamp-based versioning.

Generates a unique timestamp version (YYYYMMDDHHmmss format) to avoid conflicts when multiple developers create migrations concurrently.

Parameters:

message¶ (str) -- Description for the migration.
file_type¶ (str | None) -- Type of migration file to create ('sql' or 'py').

Return type:

None

squash(start_version=None, end_version=None, description=None, *, dry_run=False, update_database=True, yes=False, allow_gaps=False, output_format='sql')[source]¶

Squash a range of migrations into a single file.

Combines multiple sequential migrations into a single "release" migration. UP statements are merged in version order, DOWN statements in reverse order.

Parameters:

start_version¶ (str | None) -- First version in the range to squash (inclusive). When None, defaults to the first sequential migration found.
end_version¶ (str | None) -- Last version in the range to squash (inclusive). When None, defaults to the last sequential migration found.
description¶ (str | None) -- Description for the squashed migration file. When None, prompts interactively.
dry_run¶ (bool) -- Preview changes without applying.
update_database¶ (bool) -- Update migration records in database.
yes¶ (bool) -- Skip confirmation prompt.
allow_gaps¶ (bool) -- Allow gaps in version sequence.
output_format¶ (str) -- Output format ("sql" or "py").

Return type:

None

fix(dry_run=False, update_database=True, yes=False)[source]¶

Convert timestamp migrations to sequential format.

Implements hybrid versioning workflow where development uses timestamps and production uses sequential numbers. Creates backup before changes and provides rollback on errors.

Parameters:

dry_run¶ (bool) -- Preview changes without applying.
update_database¶ (bool) -- Update migration records in database.
yes¶ (bool) -- Skip confirmation prompt.

Return type:

None

class sqlspec.migrations.AsyncMigrationCommands[source]¶

Bases: BaseMigrationCommands[AsyncConfigT, Any]

Asynchronous migration commands.

__init__(config)[source]¶

Initialize migration commands.

Parameters:: config¶ (TypeVar(AsyncConfigT, bound= AsyncDatabaseConfig[Any, Any, Any] | NoPoolAsyncConfig[Any, Any])) -- The SQLSpec configuration.

async init(directory, package=True)[source]¶

Initialize migration directory structure.

Parameters:

directory¶ (str) -- Directory path for migrations.
package¶ (bool) -- Whether to create __init__.py in the directory.

Return type:

None

async current(verbose=False)[source]¶

Show current migration version.

Parameters:: verbose¶ (bool) -- Whether to show detailed migration history.
Return type:: str | None
Returns:: The current migration version or None if no migrations applied.

async upgrade(revision='head', allow_missing=False, auto_sync=True, dry_run=False, *, use_logger=False, echo=None, summary_only=None)[source]¶

Upgrade to a target revision.

Validates migration order and warns if out-of-order migrations are detected. Out-of-order migrations can occur when branches merge in different orders across environments.

Parameters:

revision¶ (str) -- Target revision or "head" for latest.
allow_missing¶ (bool) -- If True, allow out-of-order migrations even in strict mode. Defaults to False.
auto_sync¶ (bool) -- If True, automatically reconcile renamed migrations in database. Defaults to True. Can be disabled via --no-auto-sync flag.
dry_run¶ (bool) -- If True, show what would be done without making changes.
use_logger¶ (bool) -- If True, output to logger instead of Rich console. Defaults to False. Can be set via MigrationConfig for persistent default.
echo¶ (bool | None) -- Echo output to the console. Defaults to True when unset.
summary_only¶ (bool | None) -- Emit a single summary log entry when logger output is enabled.

Return type:

None

async downgrade(revision='-1', *, dry_run=False, use_logger=False, echo=None, summary_only=None)[source]¶

Downgrade to a target revision.

Parameters:

revision¶ (str) -- Target revision or "-1" for one step back.
dry_run¶ (bool) -- If True, show what would be done without making changes.
use_logger¶ (bool) -- If True, output to logger instead of Rich console. Defaults to False. Can be set via MigrationConfig for persistent default.
echo¶ (bool | None) -- Echo output to the console. Defaults to True when unset.
summary_only¶ (bool | None) -- Emit a single summary log entry when logger output is enabled.

Return type:

None

async stamp(revision)[source]¶

Mark database as being at a specific revision without running migrations.

Parameters:: revision¶ (str) -- The revision to stamp.
Return type:: None

async revision(message, file_type=None)[source]¶

Create a new migration file with timestamp-based versioning.

Generates a unique timestamp version (YYYYMMDDHHmmss format) to avoid conflicts when multiple developers create migrations concurrently.

Parameters:

message¶ (str) -- Description for the migration.
file_type¶ (str | None) -- Type of migration file to create ('sql' or 'py').

Return type:

None

async squash(start_version=None, end_version=None, description=None, *, dry_run=False, update_database=True, yes=False, allow_gaps=False, output_format='sql')[source]¶

Squash a range of migrations into a single file.

Combines multiple sequential migrations into a single "release" migration. UP statements are merged in version order, DOWN statements in reverse order.

Parameters:

start_version¶ (str | None) -- First version in the range to squash (inclusive). When None, defaults to the first sequential migration found.
end_version¶ (str | None) -- Last version in the range to squash (inclusive). When None, defaults to the last sequential migration found.
description¶ (str | None) -- Description for the squashed migration file. When None, prompts interactively.
dry_run¶ (bool) -- Preview changes without applying.
update_database¶ (bool) -- Update migration records in database.
yes¶ (bool) -- Skip confirmation prompt.
allow_gaps¶ (bool) -- Allow gaps in version sequence.
output_format¶ (str) -- Output format ("sql" or "py").

Return type:

None

async fix(dry_run=False, update_database=True, yes=False)[source]¶

Convert timestamp migrations to sequential format.

Implements hybrid versioning workflow where development uses timestamps and production uses sequential numbers. Creates backup before changes and provides rollback on errors.

Parameters:

dry_run¶ (bool) -- Preview changes without applying.
update_database¶ (bool) -- Update migration records in database.
yes¶ (bool) -- Skip confirmation prompt.

Return type:

None

sqlspec.migrations.commands.create_migration_commands(config)[source]¶

Factory function to create the appropriate migration commands.

Parameters:: config¶ -- The SQLSpec configuration.
Returns:: Appropriate migration commands instance.

Runners¶

class sqlspec.migrations.SyncMigrationRunner[source]¶

Bases: BaseMigrationRunner

Synchronous migration runner with pure sync methods.

get_migration_files()[source]¶

Get all migration files sorted by version.

Return type:: list[tuple[str, Path]]
Returns:: List of (version, path) tuples sorted by version.

load_migration(file_path, version=None)[source]¶

Load a migration file and extract its components.

Parameters:

file_path¶ (Path) -- Path to the migration file.
version¶ (str | None) -- Optional pre-extracted version (preserves prefixes like ext_adk_0001).

Return type:

dict[str, typing.Any]

Returns:

Dictionary containing migration metadata and queries.

execute_upgrade(driver, migration, *, use_transaction=None, on_success=None)[source]¶

Execute an upgrade migration.

Parameters:

driver¶ -- The sync database driver to use.
migration¶ -- Migration metadata dictionary.
use_transaction¶ -- Override transaction behavior. If None, uses should_use_transaction logic.
on_success¶ -- Callback invoked with execution_time_ms before commit (for version tracking).

Returns:

Tuple of (sql_content, execution_time_ms).

execute_downgrade(driver, migration, *, use_transaction=None, on_success=None)[source]¶

Execute a downgrade migration.

Parameters:

driver¶ -- The sync database driver to use.
migration¶ -- Migration metadata dictionary.
use_transaction¶ -- Override transaction behavior. If None, uses should_use_transaction logic.
on_success¶ -- Callback invoked with execution_time_ms before commit (for version tracking).

Returns:

Tuple of (sql_content, execution_time_ms).

load_all_migrations()[source]¶

Load all migrations into a single namespace for bulk operations.

Return type:: dict[str, SQL]
Returns:: Dictionary mapping query names to SQL objects.

class sqlspec.migrations.AsyncMigrationRunner[source]¶

Bases: BaseMigrationRunner

Asynchronous migration runner with pure async methods.

async get_migration_files()[source]¶

Get all migration files sorted by version.

Return type:: list[tuple[str, Path]]
Returns:: List of (version, path) tuples sorted by version.

async load_migration(file_path, version=None)[source]¶

Load a migration file and extract its components.

Parameters:

file_path¶ (Path) -- Path to the migration file.
version¶ (str | None) -- Optional pre-extracted version (preserves prefixes like ext_adk_0001).

Return type:

dict[str, typing.Any]

Returns:

Dictionary containing migration metadata and queries.

async execute_upgrade(driver, migration, *, use_transaction=None, on_success=None)[source]¶

Execute an upgrade migration.

Parameters:

driver¶ -- The async database driver to use.
migration¶ -- Migration metadata dictionary.
use_transaction¶ -- Override transaction behavior. If None, uses should_use_transaction logic.
on_success¶ -- Async callback invoked with execution_time_ms before commit (for version tracking).

Returns:

Tuple of (sql_content, execution_time_ms).

async execute_downgrade(driver, migration, *, use_transaction=None, on_success=None)[source]¶

Execute a downgrade migration.

Parameters:

driver¶ -- The async database driver to use.
migration¶ -- Migration metadata dictionary.
use_transaction¶ -- Override transaction behavior. If None, uses should_use_transaction logic.
on_success¶ -- Async callback invoked with execution_time_ms before commit (for version tracking).

Returns:

Tuple of (sql_content, execution_time_ms).

async load_all_migrations()[source]¶

Load all migrations into a single namespace for bulk operations.

Return type:: dict[str, SQL]
Returns:: Dictionary mapping query names to SQL objects.

sqlspec.migrations.runner.create_migration_runner(migrations_path, extension_migrations, context, extension_configs, is_async=False, runtime=None, description_hints=None)[source]¶

Overloads:

migrations_path (Path), extension_migrations (dict[str, Path]), context (MigrationContext | None), extension_configs (dict[str, Any]), is_async (Literal[False]), runtime (ObservabilityRuntime | None), description_hints (TemplateDescriptionHints | None) → SyncMigrationRunner
migrations_path (Path), extension_migrations (dict[str, Path]), context (MigrationContext | None), extension_configs (dict[str, Any]), is_async (Literal[True]), runtime (ObservabilityRuntime | None), description_hints (TemplateDescriptionHints | None) → AsyncMigrationRunner

Factory function to create the appropriate migration runner.

Parameters:

migrations_path¶ (Path) -- Path to migrations directory.
extension_migrations¶ (dict[str, Path]) -- Extension migration paths.
context¶ (MigrationContext | None) -- Migration context.
extension_configs¶ (dict[str, typing.Any]) -- Extension configurations.
is_async¶ (bool) -- Whether to create async or sync runner.
runtime¶ (ObservabilityRuntime | None) -- Observability runtime shared with loaders and execution steps.
description_hints¶ (TemplateDescriptionHints | None) -- Optional description extraction hints from template profiles.

Returns:

Appropriate migration runner instance.

Trackers¶

class sqlspec.migrations.SyncMigrationTracker[source]¶

Bases: BaseMigrationTracker[SyncDriverAdapterBase]

Synchronous migration version tracker.

ensure_tracking_table(driver)[source]¶

Create the migration tracking table if it doesn't exist.

Also checks for and adds any missing columns to support schema migrations.

Parameters:: driver¶ (SyncDriverAdapterBase) -- The database driver to use.
Return type:: None

get_current_version(driver)[source]¶

Get the latest applied migration version.

Parameters:: driver¶ (SyncDriverAdapterBase) -- The database driver to use.
Return type:: str | None
Returns:: The current version number or None if no migrations applied.

get_applied_migrations(driver)[source]¶

Get all applied migrations in order.

Parameters:: driver¶ (SyncDriverAdapterBase) -- The database driver to use.
Return type:: list[dict[str, typing.Any]]
Returns:: List of migration records.

record_migration(driver, version, description, execution_time_ms, checksum)[source]¶

Record a successfully applied migration.

Parses version to determine type (sequential or timestamp) and auto-increments execution_sequence for application order tracking.

Parameters:

driver¶ (SyncDriverAdapterBase) -- The database driver to use.
version¶ (str) -- Version number of the migration.
description¶ (str) -- Description of the migration.
execution_time_ms¶ (int) -- Execution time in milliseconds.
checksum¶ (str) -- MD5 checksum of the migration content.

Return type:

None

remove_migration(driver, version)[source]¶

Remove a migration record (used during downgrade).

Parameters:

driver¶ (SyncDriverAdapterBase) -- The database driver to use.
version¶ (str) -- Version number to remove.

Return type:

None

update_version_record(driver, old_version, new_version)[source]¶

Update migration version record from timestamp to sequential.

Updates version_num and version_type while preserving execution_sequence, applied_at, and other tracking metadata. Used during fix command.

Idempotent: If the version is already updated, logs and continues without error. This allows fix command to be safely re-run after pulling changes.

Parameters:

driver¶ (SyncDriverAdapterBase) -- The database driver to use.
old_version¶ (str) -- Current timestamp version string.
new_version¶ (str) -- New sequential version string.

Raises:

ValueError -- If neither old_version nor new_version found in database.

Return type:

None

replace_with_squash(driver, squashed_version, replaced_versions, description, checksum)[source]¶

Replace multiple migration records with a single squashed record.

Deletes all replaced version records and inserts a new record for the squashed migration with metadata about which versions it replaces.

Parameters:

driver¶ (SyncDriverAdapterBase) -- The database driver to use.
squashed_version¶ (str) -- Version number of the squashed migration.
replaced_versions¶ (list[str]) -- List of version strings being replaced.
description¶ (str) -- Description of the squashed migration.
checksum¶ (str) -- MD5 checksum of the squashed migration content.

Return type:

None

is_squash_already_applied(driver, squashed_version, replaced_versions)[source]¶

Check if a squash operation has already been applied.

Determines if any of the replaced versions exist in the database, indicating that the original migrations were applied before the squash.

Parameters:

driver¶ (SyncDriverAdapterBase) -- The database driver to use.
squashed_version¶ (str) -- Version number of the squashed migration (unused but kept for API consistency).
replaced_versions¶ (list[str]) -- List of version strings that would be replaced.

Return type:

bool

Returns:

True if any replaced version exists (squash already applied), False otherwise.

class sqlspec.migrations.AsyncMigrationTracker[source]¶

Bases: BaseMigrationTracker[AsyncDriverAdapterBase]

Asynchronous migration version tracker.

async ensure_tracking_table(driver)[source]¶

Create the migration tracking table if it doesn't exist.

Also checks for and adds any missing columns to support schema migrations.

Parameters:: driver¶ (AsyncDriverAdapterBase) -- The database driver to use.
Return type:: None

async get_current_version(driver)[source]¶

Get the latest applied migration version.

Parameters:: driver¶ (AsyncDriverAdapterBase) -- The database driver to use.
Return type:: str | None
Returns:: The current version number or None if no migrations applied.

async get_applied_migrations(driver)[source]¶

Get all applied migrations in order.

Parameters:: driver¶ (AsyncDriverAdapterBase) -- The database driver to use.
Return type:: list[dict[str, typing.Any]]
Returns:: List of migration records.

async record_migration(driver, version, description, execution_time_ms, checksum)[source]¶

Record a successfully applied migration.

Parses version to determine type (sequential or timestamp) and auto-increments execution_sequence for application order tracking.

Parameters:

driver¶ (AsyncDriverAdapterBase) -- The database driver to use.
version¶ (str) -- Version number of the migration.
description¶ (str) -- Description of the migration.
execution_time_ms¶ (int) -- Execution time in milliseconds.
checksum¶ (str) -- MD5 checksum of the migration content.

Return type:

None

async remove_migration(driver, version)[source]¶

Remove a migration record (used during downgrade).

Parameters:

driver¶ (AsyncDriverAdapterBase) -- The database driver to use.
version¶ (str) -- Version number to remove.

Return type:

None

async update_version_record(driver, old_version, new_version)[source]¶

Update migration version record from timestamp to sequential.

Updates version_num and version_type while preserving execution_sequence, applied_at, and other tracking metadata. Used during fix command.

Idempotent: If the version is already updated, logs and continues without error. This allows fix command to be safely re-run after pulling changes.

Parameters:

driver¶ (AsyncDriverAdapterBase) -- The database driver to use.
old_version¶ (str) -- Current timestamp version string.
new_version¶ (str) -- New sequential version string.

Raises:

ValueError -- If neither old_version nor new_version found in database.

Return type:

None

async replace_with_squash(driver, squashed_version, replaced_versions, description, checksum)[source]¶

Replace multiple migration records with a single squashed record.

Deletes all replaced version records and inserts a new record for the squashed migration with metadata about which versions it replaces.

Parameters:

driver¶ (AsyncDriverAdapterBase) -- The database driver to use.
squashed_version¶ (str) -- Version number of the squashed migration.
replaced_versions¶ (list[str]) -- List of version strings being replaced.
description¶ (str) -- Description of the squashed migration.
checksum¶ (str) -- MD5 checksum of the squashed migration content.

Return type:

None

async is_squash_already_applied(driver, squashed_version, replaced_versions)[source]¶

Check if a squash operation has already been applied.

Determines if any of the replaced versions exist in the database, indicating that the original migrations were applied before the squash.

Parameters:

driver¶ (AsyncDriverAdapterBase) -- The database driver to use.
squashed_version¶ (str) -- Version number of the squashed migration (unused but kept for API consistency).
replaced_versions¶ (list[str]) -- List of version strings that would be replaced.

Return type:

bool

Returns:

True if any replaced version exists (squash already applied), False otherwise.

Loaders¶

class sqlspec.migrations.BaseMigrationLoader[source]¶

Bases: ABC

Abstract base class for migration loaders.

abstractmethod async get_up_sql(path)[source]¶

Load and return the 'up' SQL statements from a migration file.

Parameters:: path¶ (Path) -- Path to the migration file.
Return type:: list[str]
Returns:: List of SQL statements to execute for upgrade.
Raises:: MigrationLoadError -- If loading fails.

abstractmethod async get_down_sql(path)[source]¶

Load and return the 'down' SQL statements from a migration file.

Parameters:: path¶ (Path) -- Path to the migration file.
Return type:: list[str]
Returns:: List of SQL statements to execute for downgrade. Empty list if no downgrade is available.
Raises:: MigrationLoadError -- If loading fails.

abstractmethod validate_migration_file(path)[source]¶

Validate that the migration file has required components.

Parameters:: path¶ (Path) -- Path to the migration file.
Raises:: MigrationLoadError -- If validation fails.
Return type:: None

class sqlspec.migrations.SQLFileLoader[source]¶

Bases: BaseMigrationLoader

Loader for SQL migration files.

__init__(sql_loader=None)[source]¶

Initialize SQL file loader.

Parameters:: sql_loader¶ (SQLFileLoader | None) -- Optional shared SQLFileLoader instance to reuse. If not provided, creates a new instance.

async get_up_sql(path)[source]¶

Extract the 'up' SQL from a SQL migration file.

The SQL file must already be loaded via validate_migration_file() before calling this method. This design ensures the file is loaded exactly once during the migration process.

Parameters:: path¶ (Path) -- Path to SQL migration file.
Return type:: list[str]
Returns:: List containing single SQL statement for upgrade.
Raises:: MigrationLoadError -- If migration file is invalid or missing up query.

async get_down_sql(path)[source]¶

Extract the 'down' SQL from a SQL migration file.

The SQL file must already be loaded via validate_migration_file() before calling this method. This design ensures the file is loaded exactly once during the migration process.

Parameters:: path¶ (Path) -- Path to SQL migration file.
Return type:: list[str]
Returns:: List containing single SQL statement for downgrade, or empty list.

validate_migration_file(path)[source]¶

Validate SQL migration file has required up query.

Parameters:: path¶ (Path) -- Path to SQL migration file.
Raises:: MigrationLoadError -- If file is invalid or missing required query.
Return type:: None

class sqlspec.migrations.PythonFileLoader[source]¶

Bases: BaseMigrationLoader

Loader for Python migration files.

__init__(migrations_dir, project_root=None, context=None)[source]¶

Initialize Python file loader.

Parameters:

migrations_dir¶ (Path) -- Directory containing migration files.
project_root¶ (Path | None) -- Optional project root directory for imports.
context¶ (Optional[typing.Any]) -- Optional migration context to pass to functions.

async get_up_sql(path)[source]¶

Load Python migration and execute upgrade function.

Parameters:: path¶ (Path) -- Path to Python migration file.
Return type:: list[str]
Returns:: List of SQL statements for upgrade.
Raises:: MigrationLoadError -- If function is missing or execution fails.

async get_down_sql(path)[source]¶

Load Python migration and execute downgrade function.

Parameters:: path¶ (Path) -- Path to Python migration file.
Return type:: list[str]
Returns:: List of SQL statements for downgrade, or empty list if not available.

validate_migration_file(path)[source]¶

Validate Python migration file has required upgrade function.

Parameters:: path¶ (Path) -- Path to Python migration file.
Raises:: MigrationLoadError -- If validation fails.
Return type:: None

sqlspec.migrations.loaders.get_migration_loader(file_path, migrations_dir, project_root=None, context=None, sql_loader=None)[source]¶

Factory function to get appropriate loader for migration file.

Parameters:

file_path¶ (Path) -- Path to the migration file.
migrations_dir¶ (Path) -- Directory containing migration files.
project_root¶ (Path | None) -- Optional project root directory for Python imports.
context¶ (Optional[typing.Any]) -- Optional migration context to pass to Python migrations.
sql_loader¶ (SQLFileLoader | None) -- Optional shared SQLFileLoader instance for SQL migrations. When provided, SQL files are loaded using this shared instance, avoiding redundant file parsing.

Return type:

BaseMigrationLoader

Returns:

Appropriate loader instance for the file type.

Raises:

MigrationLoadError -- If file type is not supported.

Squashing¶

class sqlspec.migrations.MigrationSquasher[source]¶

Bases: object

Core squash engine for combining migrations.

Provides functionality to plan, validate, and execute migration squash operations. Combines multiple sequential migrations into a single file with merged UP/DOWN SQL.

__init__(migrations_path, runner, template_settings=None)[source]¶

Initialize the migration squasher.

Parameters:

migrations_path¶ (Path) -- Path to the migrations directory.
runner¶ (SyncMigrationRunner) -- SyncMigrationRunner instance for loading migrations.
template_settings¶ (MigrationTemplateSettings | None) -- Optional template settings for generating squashed file.

plan_squash(start_version, end_version, description, *, allow_gaps=False, output_format='sql')[source]¶

Plan a squash operation for a range of migrations.

For homogeneous migrations (all SQL or all Python), returns a single plan. For mixed SQL/Python migrations, returns multiple plans - one per consecutive group of same-type migrations.

Parameters:

start_version¶ (str) -- First version in the range to squash (inclusive).
end_version¶ (str) -- Last version in the range to squash (inclusive).
description¶ (str) -- Description for the squashed migration file.
allow_gaps¶ (bool) -- If True, allow gaps in version sequence.
output_format¶ (str) -- Output file format ("sql" or "py").

Return type:

list[SquashPlan]

Returns:

List of SquashPlan objects with details of planned operations.

extract_sql(migrations)[source]¶

Extract UP and DOWN SQL statements from migrations.

UP statements are accumulated in version order. DOWN statements are accumulated in REVERSE version order for proper rollback.

Parameters:: migrations¶ (list[tuple[str, Path]]) -- List of (version, path) tuples to extract SQL from.
Return type:: tuple[list[str], list[str]]
Returns:: Tuple of (up_statements, down_statements) lists.

generate_squashed_content(plan, up_sql, down_sql)[source]¶

Generate the content for a squashed migration file.

Parameters:

plan¶ (SquashPlan) -- The SquashPlan describing the squash operation.
up_sql¶ (list[str]) -- List of UP SQL statements (in execution order).
down_sql¶ (list[str]) -- List of DOWN SQL statements (in rollback order).

Return type:

str

Returns:

Complete SQL file content as a string.

generate_python_squash(plan, up_sql, down_sql)[source]¶

Generate Python migration file content instead of SQL.

Creates a Python migration file with up() and down() functions that return the SQL statements as lists.

Parameters:

plan¶ (SquashPlan) -- The SquashPlan describing the squash operation.
up_sql¶ (list[str]) -- List of UP SQL statements (in execution order).
down_sql¶ (list[str]) -- List of DOWN SQL statements (in rollback order).

Return type:

str

Returns:

Complete Python file content as a string.

apply_squash(plans, *, dry_run=False)[source]¶

Apply the squash operation for one or more plans.

Creates backup, writes squashed files, deletes source migrations, and cleans up backup on success. Rolls back on error.

Parameters:

plans¶ (list[SquashPlan]) -- List of SquashPlan objects to execute.
dry_run¶ (bool) -- If True, no files are modified (preview only).

Return type:

None

class sqlspec.migrations.SquashPlan[source]¶

Bases: object

Represents a planned squash operation.

source_migrations¶: List of (version, path) tuples for migrations being squashed.

target_version¶: The version string for the squashed migration.

target_path¶: Output file path for the squashed migration.

description¶: Combined description for the squashed migration.

source_versions¶: List of version strings being replaced (for tracking table updates).

__init__(source_migrations, target_version, target_path, description, source_versions)¶

Version Management¶

final class sqlspec.migrations.version.MigrationVersion[source]¶

Bases: object

Parsed migration version with structured comparison support.

raw¶: Original version string.

type¶: Version format type (sequential or timestamp).

sequence¶: Numeric value for sequential versions.

timestamp¶: Parsed datetime for timestamp versions (UTC).

extension¶: Extension name for extension-prefixed versions.

__lt__(other)[source]¶

Compare versions supporting mixed formats.

Comparison Rules:

Extension migrations sort by extension name first, then version
Sequential < Timestamp (legacy migrations first)
Sequential vs Sequential: numeric comparison
Timestamp vs Timestamp: chronological comparison

Parameters:: other¶ (MigrationVersion) -- Version to compare against.
Return type:: bool
Returns:: True if this version sorts before other.

__le__(other)[source]¶

Check if version is less than or equal to another.

Parameters:: other¶ (MigrationVersion) -- Version to compare against.
Return type:: bool
Returns:: True if this version is less than or equal to other.

__gt__(other)[source]¶

Check if version is greater than another.

Parameters:: other¶ (MigrationVersion) -- Version to compare against.
Return type:: bool
Returns:: True if this version sorts after other.

__ge__(other)[source]¶

Check if version is greater than or equal to another.

Parameters:: other¶ (MigrationVersion) -- Version to compare against.
Return type:: bool
Returns:: True if this version is greater than or equal to other.

__eq__(other)[source]¶

Check version equality.

Parameters:: other¶ (object) -- Version to compare against.
Return type:: bool
Returns:: True if versions are equal.

__hash__()[source]¶

Hash version for use in sets and dicts.

Return type:: int
Returns:: Hash value based on raw version string.

__repr__()[source]¶

Get string representation for debugging.

Return type:: str
Returns:: String representation with type and value.

__init__(raw, type, sequence, timestamp, extension)¶

class sqlspec.migrations.version.VersionType[source]¶

Bases: Enum

Migration version format type.

Context¶

class sqlspec.migrations.context.MigrationContext[source]¶

Bases: object

Context object passed to migration functions.

Provides runtime information about the database environment to migration functions, allowing them to generate dialect-specific SQL.

config: Any | None¶: Database configuration object.

dialect: str | None¶: Database dialect.

metadata: dict[str, Any] | None¶: Additional metadata for the migration.

extension_config: dict[str, Any] | None¶: Extension-specific configuration options.

driver: SyncDriverAdapterBase | AsyncDriverAdapterBase | None¶: Database driver instance (available during execution).

__post_init__()[source]¶

Initialize metadata and extension config if not provided.

Return type:: None

classmethod from_config(config)[source]¶

Create context from database configuration.

Parameters:: config¶ (Any) -- Database configuration object.
Return type:: MigrationContext
Returns:: Migration context with dialect information.

property is_async_execution: bool¶

Check if migrations are running in an async execution context.

Returns:: True if executing in an async context.

__init__(config=None, dialect=None, metadata=None, extension_config=None, driver=None, _execution_metadata=<factory>)¶

property is_async_driver: bool¶

Check if the current driver is async.

Returns:: True if driver supports async operations.

property execution_mode: str¶

Get the current execution mode.

Returns:: 'async' if in async context, 'sync' otherwise.

set_execution_metadata(key, value)[source]¶

Set execution metadata for tracking migration state.

Parameters:

key¶ (str) -- Metadata key.
value¶ (Any) -- Metadata value.

Return type:

None

get_execution_metadata(key, default=None)[source]¶

Get execution metadata.

Parameters:

key¶ (str) -- Metadata key.
default¶ (Any) -- Default value if key not found.

Return type:

Any

Returns:

Metadata value or default.

validate_async_usage(migration_func)[source]¶

Validate proper usage of async functions in migration context.

Parameters:: migration_func¶ (Any) -- The migration function to validate.
Return type:: None

Fixer¶

class sqlspec.migrations.fix.MigrationFixer[source]¶

Bases: object

Handles atomic migration file conversion operations.

Provides backup/rollback functionality and manages conversion from timestamp-based migration files to sequential format.

__init__(migrations_path)[source]¶

Initialize migration fixer.

Parameters:: migrations_path¶ (Path) -- Path to migrations directory.

plan_renames(conversion_map)[source]¶

Plan all file rename operations from conversion map.

Scans migration directory and builds list of MigrationRename objects for all files that need conversion. Validates no target collisions.

Parameters:: conversion_map¶ (dict[str, str]) -- Dictionary mapping old versions to new versions.
Return type:: list[MigrationRename]
Returns:: List of planned rename operations.
Raises:: ValueError -- If target file already exists or collision detected.

create_backup()[source]¶

Create timestamped backup directory with all migration files.

Keep in sync with MigrationSquasher._create_backup.

Return type:: Path
Returns:: Path to created backup directory.

apply_renames(renames, dry_run=False)[source]¶

Execute planned rename operations.

Parameters:

renames¶ (list[MigrationRename]) -- List of planned rename operations.
dry_run¶ (bool) -- If True, log operations without executing.

Return type:

None

update_file_content(file_path, old_version, new_version)[source]¶

Update SQL query names and version comments in file content.

Transforms query names and version metadata from old version to new version:: -- name: migrate-{old_version}-up → -- name: migrate-{new_version}-up -- name: migrate-{old_version}-down → -- name: migrate-{new_version}-down -- Version: {old_version} → -- Version: {new_version}

Creates version-specific regex patterns to avoid unintended replacements of other migrate-* patterns in the file.

Parameters:

file_path¶ (Path) -- Path to file to update.
old_version¶ (str | None) -- Old version string (None values skipped gracefully).
new_version¶ (str | None) -- New version string (None values skipped gracefully).

Return type:

None

rollback()[source]¶

Restore migration files from backup.

Keep in sync with MigrationSquasher._rollback_backup.

Deletes current migration files and restores from backup directory. Only restores if backup exists.

Return type:: None

cleanup()[source]¶

Remove backup directory after successful conversion.

Keep in sync with MigrationSquasher._cleanup_backup.

Only removes backup if it exists. Logs warning if no backup found.

Return type:: None

class sqlspec.migrations.fix.MigrationRename[source]¶

Bases: object

Represents a planned migration file rename operation.

old_path¶: Current file path.

new_path¶: Target file path after rename.

old_version¶: Current version string.

new_version¶: Target version string.

needs_content_update¶: Whether file content needs updating. True for SQL files that contain query names.

__init__(old_path, new_path, old_version, new_version, needs_content_update)¶

Templates¶

class sqlspec.migrations.templates.MigrationTemplateSettings[source]¶

Bases: object

Resolved template configuration for a migration command context.

resolve_format(requested)[source]¶

Resolve the effective file format to render.

Return type:: str

property description_hints: TemplateDescriptionHints¶: Expose description extraction hints derived from the active profile.

__init__(default_format, profile)¶

class sqlspec.migrations.templates.MigrationTemplateProfile[source]¶

Bases: object

Concrete template profile selected via configuration.

__init__(name, title, sql, python)¶

class sqlspec.migrations.templates.SQLTemplateDefinition[source]¶

Bases: object

SQL migration template fragments.

render(context)[source]¶

Render the SQL template using the supplied context.

Return type:: str

__init__(header, metadata=<factory>, body='', description_keys=('Description', ))¶

class sqlspec.migrations.templates.PythonTemplateDefinition[source]¶

Bases: object

Python migration template fragments.

render(context)[source]¶

Render the Python template using the supplied context.

Return type:: str

__init__(docstring, body, imports=<factory>, description_keys=('Description', ))¶

Utilities¶

sqlspec.migrations.create_migration_file(migrations_dir, version, message, file_type=None, *, config=None, template_settings=None)[source]¶

Create a new migration file from template.

Return type:: Path

async sqlspec.migrations.drop_all(engine, version_table_name, metadata=None)[source]¶

Drop all tables from the database.

Parameters:

engine¶ (AsyncDriverAdapterBase) -- The database engine/driver.
version_table_name¶ (str) -- Name of the version tracking table.
metadata¶ (Any | None) -- Optional metadata object.

Raises:

NotImplementedError -- Always raised.

Return type:

None

sqlspec.migrations.get_author(author_config=None, *, config=None)[source]¶

Resolve author metadata for migration templates.

Return type:: str