Issue
Django migrations has excellent behavior in terms of single migrations, where, assuming you leave atomic=True
, a migration will be all-or-nothing: it will either run to completion or undo everything.
Is there a way to get this all-or-nothing behavior for multiple migrations? That is to say, is there a way to either run multiple migrations inside an enclosing transaction (which admittedly could cause other problems) or to rollback all succeeded migrations on failure?
For context, I'm looking for a single command or setting to do this so that I can include it in a deploy script. Currently, the only part of my deploy that isn't rolled back in the event of a failure are database changes. I know that this can be manually done by running python manage.py migrate APP_NAME MIGRATION_NUMBER
in the event of a failure, but this requires knowledge of the last run migration on each app.
Solution
This isn't a feature in Django (at least as of 4.1). To do this yourself, your code needs to do the following:
- Get a list of current apps that have migrations. This can be done by getting the intersection of apps from settings.INSTALLED_APPS and the apps in the MigrationRecorder.
- Get the latest migration for each app.
- Run migrations.
- If migrations fail, run the rollback command for the latest migration for each app.
Here's an implementation of this as a custom management command. This is done as a child-class of the migrate management command so that it exposes all of the command line options of migrate. To use this, put it in a file called <app>/management/commands/migrate_or_rollback.py
(where <app>
is any of your installed apps) and you can invoke it with python manage.py migrate_or_rollback
:
from django.db.migrations.recorder import MigrationRecorder
from django.core.management import call_command
from django.conf import settings
from django.core.management.commands import migrate
from pprint import pprint
class Command(migrate.Command):
def handle(self, *args, **options):
installed_apps = [i.split(".")[-1] for i in settings.INSTALLED_APPS]
migration_apps = [
i["app"] for i in MigrationRecorder.Migration.objects.values("app").distinct()
]
migration_apps = [i for i in migration_apps if i in installed_apps]
latest_migration_by_app = {
app: MigrationRecorder.Migration.objects.filter(app=app).latest("id").name
for app in migration_apps
}
try:
super().handle(*args, **options)
except Exception:
print("\n\nMigrating failed; rolling back to last migration state:\n")
pprint(latest_migration_by_app)
print()
for app, latest_migration in latest_migration_by_app.items():
call_command("migrate", app, latest_migration)
print("\nRollboack successfull\n")
raise
Answered By - Zags
0 comments:
Post a Comment
Note: Only a member of this blog may post a comment.