type_bridge.migration.loader

loader

Migration file loader and discovery.

Discovers and loads migration files from a directory structure. Migration files must follow the naming convention: NNNN_name.py (e.g., 0001_initial.py)

LoadedMigration dataclass

LoadedMigration(migration, path, checksum)

A migration loaded from a file.

Attributes:

Name	Type	Description
migration	Migration	The Migration instance
path	Path	Path to the migration file
checksum	str	SHA256 hash of file content (first 16 chars)

MigrationLoadError

Bases: Exception

Error loading a migration file.

MigrationLoader

MigrationLoader(migrations_dir)

Loads migration files from a directory.

Migration files must follow the naming pattern: NNNN_*.py where NNNN is a 4-digit number (e.g., 0001_initial.py, 0002_add_company.py)

Example

loader = MigrationLoader(Path("migrations")) migrations = loader.discover()

for loaded in migrations: print(f"{loaded.migration.name}: {loaded.checksum}")

Initialize loader.

Parameters:

Name	Type	Description	Default
migrations_dir	Path	Directory containing migration files	required

Source code in type_bridge/migration/loader.py

def __init__(self, migrations_dir: Path):
    """Initialize loader.

    Args:
        migrations_dir: Directory containing migration files
    """
    self.migrations_dir = migrations_dir

discover

discover()

Discover all migration files in order.

Returns:

Type	Description
list[LoadedMigration]	List of loaded migrations, sorted by filename

Source code in type_bridge/migration/loader.py

def discover(self) -> list[LoadedMigration]:
    """Discover all migration files in order.

    Returns:
        List of loaded migrations, sorted by filename
    """
    if not self.migrations_dir.exists():
        logger.debug(f"Migrations directory does not exist: {self.migrations_dir}")
        return []

    files = sorted(self.migrations_dir.glob(self.MIGRATION_PATTERN))
    migrations: list[LoadedMigration] = []

    for path in files:
        try:
            loaded = self._load_migration_file(path)
            if loaded:
                migrations.append(loaded)
        except Exception as e:
            logger.error(f"Failed to load migration {path}: {e}")
            raise MigrationLoadError(f"Failed to load migration {path}: {e}") from e

    logger.debug(f"Discovered {len(migrations)} migration(s) in {self.migrations_dir}")
    return migrations

get_by_name

get_by_name(name)

Get a specific migration by name.

Parameters:

Name	Type	Description	Default
name	str	Migration name (e.g., "0001_initial")	required

Returns:

Type	Description
LoadedMigration | None	LoadedMigration or None if not found

Source code in type_bridge/migration/loader.py

def get_by_name(self, name: str) -> LoadedMigration | None:
    """Get a specific migration by name.

    Args:
        name: Migration name (e.g., "0001_initial")

    Returns:
        LoadedMigration or None if not found
    """
    for loaded in self.discover():
        if loaded.migration.name == name:
            return loaded
    return None

get_by_number

get_by_number(number)

Get a specific migration by number.

Parameters:

Name	Type	Description	Default
number	int	Migration number (e.g., 1 for 0001_initial)	required

Returns:

Type	Description
LoadedMigration | None	LoadedMigration or None if not found

Source code in type_bridge/migration/loader.py

def get_by_number(self, number: int) -> LoadedMigration | None:
    """Get a specific migration by number.

    Args:
        number: Migration number (e.g., 1 for 0001_initial)

    Returns:
        LoadedMigration or None if not found
    """
    prefix = f"{number:04d}_"
    for loaded in self.discover():
        if loaded.migration.name.startswith(prefix):
            return loaded
    return None

get_next_number

get_next_number()

Get the next available migration number.

Returns:

Type	Description
int	Next migration number (1 if no migrations exist)

Source code in type_bridge/migration/loader.py

def get_next_number(self) -> int:
    """Get the next available migration number.

    Returns:
        Next migration number (1 if no migrations exist)
    """
    migrations = self.discover()
    if not migrations:
        return 1

    # Extract numbers from existing migrations
    numbers = []
    for loaded in migrations:
        try:
            num = int(loaded.migration.name[:4])
            numbers.append(num)
        except (ValueError, IndexError):
            pass

    return max(numbers) + 1 if numbers else 1

validate_dependencies

validate_dependencies()

Validate that all migration dependencies are satisfied.

Returns:

Type	Description
list[str]	List of error messages (empty if valid)

Source code in type_bridge/migration/loader.py

def validate_dependencies(self) -> list[str]:
    """Validate that all migration dependencies are satisfied.

    Returns:
        List of error messages (empty if valid)
    """
    migrations = self.discover()
    errors: list[str] = []

    # Build set of available migrations
    available = {(m.migration.app_label, m.migration.name) for m in migrations}

    for loaded in migrations:
        for dep_app, dep_name in loaded.migration.dependencies:
            if (dep_app, dep_name) not in available:
                errors.append(
                    f"Migration {loaded.migration.name} depends on "
                    f"{dep_app}.{dep_name} which does not exist"
                )

    return errors