Skip to content

Commit

Permalink
rebase_migration: Un apply and rebase migrations in case migrations a…
Browse files Browse the repository at this point in the history
…re applied.

Previously, while rebasing migrations, if the migration to be
rebased is already applied, it would throw out an error asking
to un apply the migration before rebasing.

This commit un applies the migrations and rebases them when
the migrations to be rebased are already applied. This is done
by first creating a --merge migration, un applying the previous
migrations, removing the --merge migration, and then rebasing
the existing migrations.

Fixes #184
  • Loading branch information
roanster007 committed Aug 5, 2024
1 parent e0b27b9 commit d537bc3
Showing 1 changed file with 61 additions and 5 deletions.
Original file line number Diff line number Diff line change
Expand Up @@ -93,11 +93,16 @@ def rebase_migration(
)

if migration_applied(app_label, rebased_migration_name):
raise CommandError(
f"Detected {rebased_migration_name} as the rebased migration,"
+ " but it is applied to the local database. Undo the rebase,"
+ " reverse the migration, and try again."
)
try:
unapply_migrations(
app_label, rebased_migration_name, self.last_migration_name
)
except (FileNotFoundError, subprocess.SubprocessError):
raise CommandError(
f"Detected {rebased_migration_name} as the rebased migration,"
+ " but it is applied to the local database. Undo the rebase,"
+ " reverse the migration, and try again."
)

content = rebased_migration_path.read_text()
split_result = re.split(
Expand Down Expand Up @@ -272,3 +277,54 @@ def migration_applied(app_label: str, migration_name: str) -> bool:
# django_migrations table does not exist -> no migrations applied
pass
return False


def unapply_migrations(
app_label: str, first_rebase_migration: str, last_merged_migration: str
) -> None: # pragma: no cover
first_rebase_migration_number, _merged_rest = first_rebase_migration.split("_", 1)
last_merged_number, _merged_rest = last_merged_migration.split("_", 1)
migration_details = MigrationDetails(app_label)

subprocess.run(
[
"./manage.py",
"makemigrations",
"--merge",
"--noinput",
"--skip-checks",
f"{app_label}",
],
check=True,
)

last_migration_to_be_applied = None
merge_migration_name = None

for migration_name in migration_details.names:
if migration_name.startswith(
f"{int(first_rebase_migration_number) - 1}".zfill(4)
):
last_migration_to_be_applied = migration_name

elif migration_name.startswith(f"{int(last_merged_number) + 1}".zfill(4)):
merge_migration_name = migration_name

assert last_migration_to_be_applied is not None and merge_migration_name is not None
merge_migration_path = migration_details.dir / f"{merge_migration_name}.py"

subprocess.run(
[
"./manage.py",
"migrate",
"--skip-checks",
f"{app_label}",
last_migration_to_be_applied,
],
check=True,
)

subprocess.run(
["rm", f"{merge_migration_path}"],
check=True,
)

0 comments on commit d537bc3

Please sign in to comment.