A script to automatically resolve migration conflicts when merging in updates from other devs.

update_migrations.py

import os
import re
import sys
import shlex
import subprocess


REPO_DIR = os.environ.get("REPO_LOCATION", None)
MIGRATIONS_DIR = os.path.join(REPO_DIR, 'core', 'migrations')
MAX_MIGRATION = os.path.join(MIGRATIONS_DIR, 'max_migration.txt')
MIGRATION_REFERENCE = re.compile(r'\"core\", \"\d{4}_.+\"')


def run_cmd(cmd: str) -> str:
    """Run a terminal command and return output."""
    run = subprocess.run(shlex.split(cmd), stdout=subprocess.PIPE, text=True)
    return run.stdout


def get_branch_migrations() -> list[str]:
    """Return migration files that have been added/changed from the develop branch."""
    raw_files = run_cmd('git diff --name-only develop...')
    files = raw_files.split()
    return sorted([
        f.split('/')[-1] for f in files
        if 'migrations' in f
        and '.py' in f
    ])


def get_migration_number(migration: str) -> int:
    """Return the number part of a migration name."""
    number, _ = migration.split('_', maxsplit=1)
    return int(number)


def get_migration_name(migration: str) -> str:
    """Return the name part of a migration name."""
    _, name = migration.split('_', maxsplit=1)
    return os.path.splitext(name)[0]


def filter_branch_migrations(migrations: list[str]):
    """Filter out any old edited migrations from being updated."""
    migrations_copy = migrations.copy()
    for idx, migration in enumerate(migrations_copy[:-1]):
        number = get_migration_number(migration)
        next_number = get_migration_number(migrations_copy[idx+1])
        if number != next_number - 1:
            print(f'Skipping {migration}: not consecutive.')
            migrations.remove(migration)


def output_branch_migrations(migrations: list[str]):
    """Print out new branch migrations."""
    print(f'You have {len(migrations)} new branch migrations:')
    for migration in migrations:
        print(f'    {migration}')
    if not migrations:
        print('There are no migrations to update.')
        sys.exit()


def get_latest_develop_migration(new_migrations: list[str]) -> str:
    """Gets latest migration from the develop branch."""
    develop_migrations = sorted([
        f for f in os.listdir(MIGRATIONS_DIR)
        if 'py' in f                 # only python files
        and '__' not in f            # exclude __init__
        and f not in new_migrations  # exclude new migrations
    ])
    return os.path.splitext(develop_migrations[-1])[0]


def update_max_migration(latest_dev_migration: str) -> list[str]:
    """Update the max_migration file with the latest migration name."""
    latest_migration_no_ext = os.path.splitext(latest_dev_migration)[0]
    print(f'Last migration: setting max_migration to {latest_migration_no_ext}')
    with open(MAX_MIGRATION, 'w') as f:
        f.write(latest_migration_no_ext + '\n')


def update_migrations(migrations: list[str]):
    """Goes through branch migrations and updates them to ensure consistent history."""
    latest_dev_migration = get_latest_develop_migration(migrations)
    latest_dev_num = get_migration_number(latest_dev_migration)
    first_branch_num = get_migration_number(migrations[0])
    changed_files = []

    print(f'Latest develop migration number: {latest_dev_num}')
    print(f'Earliest new branch migration number: {first_branch_num}')
    if first_branch_num == latest_dev_num + 1:
        print('Migrations up to date.')
        sys.exit()
    elif first_branch_num > latest_dev_num + 1:
        print('Missing migrations from develop branch - please check.')
        sys.exit()

    for migration in migrations:
        print(f'Fixing {migration}...')
        dev_number = get_migration_number(latest_dev_migration)
        branch_name = get_migration_name(migration)
        migration_filename = os.path.join(MIGRATIONS_DIR, migration)
        with open(migration_filename, 'r+') as f:
            content = f.read()
            if len(re.findall(MIGRATION_REFERENCE, content)) > 1:
                print(
                    f'Cannot automatically update {migration} - '
                    'more than one core dependency.'
                )
                sys.exit()
            new_reference_content = re.sub(
                MIGRATION_REFERENCE,
                f'"core", "{latest_dev_migration}"',
                content
            )
            f.seek(0)
            f.write(new_reference_content)
            f.truncate()

        latest_dev_migration = f'{str(dev_number + 1).zfill(4)}_{branch_name}'
        latest_dev_migration_with_ext = latest_dev_migration + '.py'
        print(f'    renaming to {latest_dev_migration_with_ext}')
        new_filename = os.path.join(MIGRATIONS_DIR, latest_dev_migration_with_ext)
        os.rename(migration_filename, new_filename)
        changed_files.append(migration_filename)
        changed_files.append(new_filename)

    update_max_migration(latest_dev_migration)
    changed_files.append(MAX_MIGRATION)
    return changed_files


def commit_changed_files(changed_files: list[str]):
    """Adds and commits the given list of files."""
    file_list = ' '.join(changed_files)
    run_cmd(f'git add {file_list}')
    run_cmd(f"git commit {file_list} -m 'Update migrations'")


if __name__ == '__main__':
    if REPO_DIR is None:
        print("Set the REPO_LOCATION environment variable before running this script")

    os.chdir(REPO_DIR)
    branch_migrations = get_branch_migrations()
    filter_branch_migrations(branch_migrations)
    output_branch_migrations(branch_migrations)
    changed_files = update_migrations(branch_migrations)
    commit_changed_files(changed_files)

A script intended to reduce the tedium involved when updating your branch if migrations are involved. You can either delete your migrations and run makemigrations again, but lose any run_python scripts you added, or you can manually rename all your migrations and change their dependencies to maintain a consistent history. This script does the latter for you.
Run this script after a merge. It will detect any clashes in dependencies, (hopefully) resolve them for you, and commit (but not push) the changes automatically. Feel free to examine the resulting commit before pushing to ensure it is correct.

Example, with an old edited migration (that should not be updated), and two new migrations (0478, 0479). After merging in develop, there are now two 0478 migrations:

Skipping 0472_old.py: not consecutive.
You have 2 new branch migrations:
    0478_test.py
    0479_test_2.py
Latest develop migration number: 478
Earliest new branch migration number: 478
Fixing 0478_test.py...
    renaming to 0479_test.py
Fixing 0479_test_2.py...
    renaming to 0480_test_2.py
Last migration: setting max_migration to 0480_test_2