alembic_converger package

Submodules

alembic_converger.cli module

Command-line interface for alembic-converger.

This module provides a Click-based CLI for converging Alembic migrations.

alembic_converger.cli.main()[source]: Entry point for the CLI.

alembic_converger.config module

Alembic configuration loading and validation.

alembic_converger.config.get_revision_map(script_dir: ScriptDirectory) → RevisionMap[source]

Get RevisionMap from ScriptDirectory.

Parameters:: script_dir – ScriptDirectory instance.
Returns:: RevisionMap instance for graph traversal.

alembic_converger.config.get_script_directory(config: Config) → ScriptDirectory[source]

Get ScriptDirectory instance from Alembic config.

Parameters:: config – Alembic Config object.
Returns:: ScriptDirectory instance.
Raises:: ConfigurationError – If script directory cannot be created.

alembic_converger.config.load_alembic_config(config_path: str | None = None) → Config[source]

Load Alembic configuration from file.

Parameters:: config_path – Path to alembic.ini file. If None, uses alembic.ini in current directory.
Returns:: Alembic Config object.
Raises:: ConfigurationError – If configuration cannot be loaded.

alembic_converger.converge module

Core convergence algorithm for Alembic migration graphs.

This module orchestrates the entire convergence process: 1. Load configuration and revision map 2. Find heads descending from base revision 3. Iteratively merge heads until only one remains 4. Validate after each merge

class alembic_converger.converge.ConvergenceResult[source]

Bases: object

Result of a convergence operation.

alembic_converger.converge.converge_migrations(base_revision: str, config_path: str | None = None, dry_run: bool = False, skip_validation: bool = False, verbose: bool = False) → ConvergenceResult[source]

Converge multiple Alembic migration heads into a single head.

This is the main entry point for the convergence algorithm.

Algorithm: 1. Load Alembic configuration and revision map 2. Find all heads descending from base_revision 3. If 0 or 1 heads, return (already converged) 4. Validate all heads share a common base 5. While more than one head exists:

Pick two heads (deterministically)

Create merge migration

Validate merge is empty (no schema ops)

Optionally validate upgrade path

Reload revision map

Return convergence result

Parameters:

base_revision – Base revision to converge from. All heads must descend from this revision.
config_path – Path to alembic.ini file. If None, uses “alembic.ini” in current directory.
dry_run – If True, show what would be done without making changes.
skip_validation – If True, skip upgrade validation (faster but less safe).
verbose – If True, enable debug logging.

Returns:

ConvergenceResult object with details about the convergence.

Raises:

ConvergenceError – If convergence fails for any reason.
ConfigurationError – If Alembic configuration is invalid.
MultipleBasesError – If heads have unrelated bases.
ValidationError – If upgrade validation fails.
SchemaConflictError – If merge contains schema operations.

alembic_converger.errors module

Custom exceptions for the alembic-converger package.

All exceptions inherit from ConvergenceError for easy catching.

exception alembic_converger.errors.ConfigurationError[source]

Bases: ConvergenceError

Raised when Alembic configuration is invalid or cannot be loaded.

exception alembic_converger.errors.ConvergenceError[source]

Bases: Exception

Base exception for all convergence-related errors.

exception alembic_converger.errors.GraphError[source]

Bases: ConvergenceError

Raised when migration graph analysis fails.

exception alembic_converger.errors.MultipleBasesError[source]

Bases: ConvergenceError

Raised when multiple unrelated base revisions are detected.

This indicates that the migration heads do not share a common ancestor, which makes automatic convergence unsafe.

exception alembic_converger.errors.SchemaConflictError[source]

Bases: ConvergenceError

Raised when a merge migration contains schema operations.

Merge migrations should only contain graph operations (down_revision tuple). Schema operations indicate conflicts that require manual resolution.

exception alembic_converger.errors.ValidationError[source]

Bases: ConvergenceError

Raised when migration upgrade validation fails.

This typically indicates that the merge migration created conflicts or that the upgrade path is broken.

alembic_converger.graph module

Migration graph analysis and traversal utilities.

This module provides functions for analyzing the Alembic migration DAG, including head detection, ancestor queries, and graph validation.

alembic_converger.graph.get_common_ancestor(revision_map: RevisionMap, rev1: str, rev2: str) → str | None[source]

Find the lowest common ancestor of two revisions.

Parameters:

revision_map – Alembic RevisionMap instance.
rev1 – First revision ID.
rev2 – Second revision ID.

Returns:

Common ancestor revision ID, or None if no common ancestor exists.

alembic_converger.graph.get_heads_from_base(revision_map: RevisionMap, base_revision: str) → List[str][source]

Find all heads that descend from the given base revision.

A head is a revision with no descendants. This function finds all heads that can be reached by traversing forward from the base revision.

Parameters:

revision_map – Alembic RevisionMap instance.
base_revision – Starting revision ID.

Returns:

List of head revision IDs that descend from base.

Raises:

GraphError – If base revision doesn’t exist or graph traversal fails.

alembic_converger.graph.get_merge_pairs(heads: List[str]) → List[Tuple[str, str]][source]

Generate pairs of heads to merge.

For efficient convergence, we merge heads pairwise until only one remains. This generates the minimal number of merge operations.

Parameters:: heads – List of head revision IDs (should be sorted deterministically).
Returns:: List of (head1, head2) tuples to merge.

alembic_converger.graph.is_descendant_of(revision_map: RevisionMap, revision: str, ancestor: str) → bool[source]

Check if revision is a descendant of ancestor.

Parameters:

revision_map – Alembic RevisionMap instance.
revision – Revision to check.
ancestor – Potential ancestor revision.

Returns:

True if revision descends from ancestor, False otherwise.

alembic_converger.graph.sort_heads_deterministically(heads: List[str]) → List[str][source]

Sort heads deterministically for reproducible convergence.

Parameters:: heads – List of head revision IDs.
Returns:: Sorted list of head revision IDs.

alembic_converger.graph.validate_single_base(revision_map: RevisionMap, heads: List[str]) → None[source]

Validate that all heads share a single common base.

This ensures that the heads can be safely merged - they must all descend from a common ancestor.

Parameters:

revision_map – Alembic RevisionMap instance.
heads – List of head revision IDs.

Raises:

MultipleBasesError – If heads have multiple unrelated bases.

alembic_converger.merge module

Merge migration creation using Alembic’s Python API.

This module handles the creation of merge migrations and validation that they contain only graph operations (no schema changes).

alembic_converger.merge.create_merge_migration(config: Config, script_dir: ScriptDirectory, heads: List[str], message: str | None = None) → str[source]

Create a merge migration for the given heads.

Uses Alembic’s command.merge() to generate a merge migration file.

Parameters:

config – Alembic Config object.
script_dir – ScriptDirectory instance.
heads – List of head revision IDs to merge (typically 2).
message – Optional merge message. If None, auto-generated.

Returns:

Revision ID of the created merge migration.

Raises:

SchemaConflictError – If the merge contains schema operations.

alembic_converger.merge.delete_merge_migration(script_dir: ScriptDirectory, revision: str) → None[source]

Delete a merge migration file (used for rollback on failure).

Parameters:

script_dir – ScriptDirectory instance.
revision – Revision ID to delete.

alembic_converger.merge.has_schema_operations(migration_content: str) → bool[source]

Check if migration content contains schema operations.

A clean merge should have empty upgrade() and downgrade() functions with only ‘pass’ statements.

Parameters:: migration_content – Content of the migration file.
Returns:: True if schema operations are found, False otherwise.

alembic_converger.merge.validate_merge_is_empty(script_dir: ScriptDirectory, revision: str) → None[source]

Validate that a merge migration contains no schema operations.

Merge migrations should only update the down_revision tuple to point to multiple parents. They should not contain any upgrade() or downgrade() operations.

Parameters:

script_dir – ScriptDirectory instance.
revision – Revision ID to validate.

Raises:

SchemaConflictError – If the merge contains schema operations.

alembic_converger.utils module

Utility functions for the alembic-converger package.

alembic_converger.utils.format_revision_id(revision: str | None) → str[source]

Format a revision ID for display.

Parameters:: revision – Revision ID or None.
Returns:: Formatted string.

alembic_converger.utils.get_logger() → Logger[source]

Get the package logger.

Returns:: Logger instance.

alembic_converger.utils.resolve_config_path(config_path: str | None = None) → Path[source]

Resolve the Alembic configuration file path.

Parameters:: config_path – Path to alembic.ini file. If None, looks for alembic.ini in the current directory.
Returns:: Resolved Path object.
Raises:: FileNotFoundError – If the configuration file doesn’t exist.

alembic_converger.utils.setup_logging(verbose: bool = False) → Logger[source]

Set up logging for the package.

Parameters:: verbose – If True, set log level to DEBUG, otherwise INFO.
Returns:: Configured logger instance.

alembic_converger.validate module

Migration validation through upgrade testing.

This module provides functions to validate that migrations can be successfully applied by running upgrade operations.

alembic_converger.validate.check_database_available(config: Config) → bool[source]

Check if database is available for validation.

Parameters:: config – Alembic Config object.
Returns:: True if database connection can be established, False otherwise.

alembic_converger.validate.validate_upgrade(config: Config, from_revision: str, to_revision: str = 'head', use_subprocess: bool = True) → None[source]

Validate that upgrade from one revision to another succeeds.

This runs an actual Alembic upgrade to test the migration path. It requires a database connection configured in alembic.ini.

Parameters:

config – Alembic Config object.
from_revision – Starting revision.
to_revision – Target revision (default: “head”).
use_subprocess – If True, use subprocess to isolate the upgrade. If False, use Alembic’s command API directly.

Raises:

ValidationError – If the upgrade fails.

Module contents

Alembic Converger - Safely converge Alembic migration graphs.

This package provides tools to automatically detect and converge multiple Alembic migration heads by generating minimal merge migrations without schema operations.

exception alembic_converger.ConfigurationError[source]

Bases: ConvergenceError

Raised when Alembic configuration is invalid or cannot be loaded.

exception alembic_converger.ConvergenceError[source]

Bases: Exception

Base exception for all convergence-related errors.

exception alembic_converger.MultipleBasesError[source]

Bases: ConvergenceError

Raised when multiple unrelated base revisions are detected.

This indicates that the migration heads do not share a common ancestor, which makes automatic convergence unsafe.

exception alembic_converger.SchemaConflictError[source]

Bases: ConvergenceError

Raised when a merge migration contains schema operations.

Merge migrations should only contain graph operations (down_revision tuple). Schema operations indicate conflicts that require manual resolution.

exception alembic_converger.ValidationError[source]

Bases: ConvergenceError

Raised when migration upgrade validation fails.

This typically indicates that the merge migration created conflicts or that the upgrade path is broken.

alembic_converger.converge_migrations(base_revision: str, config_path: str | None = None, dry_run: bool = False, skip_validation: bool = False, verbose: bool = False) → ConvergenceResult[source]

Converge multiple Alembic migration heads into a single head.

This is the main entry point for the convergence algorithm.

Algorithm: 1. Load Alembic configuration and revision map 2. Find all heads descending from base_revision 3. If 0 or 1 heads, return (already converged) 4. Validate all heads share a common base 5. While more than one head exists:

Pick two heads (deterministically)

Create merge migration

Validate merge is empty (no schema ops)

Optionally validate upgrade path

Reload revision map

Return convergence result

Parameters:

base_revision – Base revision to converge from. All heads must descend from this revision.
config_path – Path to alembic.ini file. If None, uses “alembic.ini” in current directory.
dry_run – If True, show what would be done without making changes.
skip_validation – If True, skip upgrade validation (faster but less safe).
verbose – If True, enable debug logging.

Returns:

ConvergenceResult object with details about the convergence.

Raises:

ConvergenceError – If convergence fails for any reason.
ConfigurationError – If Alembic configuration is invalid.
MultipleBasesError – If heads have unrelated bases.
ValidationError – If upgrade validation fails.
SchemaConflictError – If merge contains schema operations.