python - 在生产系统中重置 Django 迁移

Question

自从几年前开始使用 Django 迁移以来，我已经阅读了很多涉及 Django 迁移的帖子/文章/反复试验，所以我决定发布一个 self 回答的问题，注明正确的方法在生产数据库中完成迁移的干净重置，让您拥有与离开时相同的数据库结构，但从初始迁移开始全新。

总体问题是这样的:

当你有一个更大的项目时，你开始为使用 Django 构建的系统积累大量的迁移。这通常不是问题，但是当您开始累积超过 50-100 个迁移文件(其中很多是相同字段的添加和删除)时，最好有一个“清理”选项，因为它应该是众所周知，如果您错误地更改了迁移历史记录，您将留下一个系统，该系统或多或少地卡住在以前的数据库状态中，解决该问题的唯一方法是手动进行基于 sql 的迁移更改。

Answer 1

我针对这个问题提出的解决方案分步骤进行:

第一步创建迁移以删除您想要的任何模型或字段并在本地运行它们，您的开发系统必须与所有其他开发系统以及生产系统同步......如果不是这种情况，您需要确保它是!

第 2 步

1. 删除本地迁移文件(一个不错的选择是更改下面的命令，我目前有一个目录结构，我的系统应用程序位于名为 /apps/ 的目录中)

运行调用 python manage.py delete_local_migration_files(如果你这样命名的话)

import os

import django.apps
from django.conf import settings
from django.core.management.base import BaseCommand


def delete_migrations(app):
    print(f"Deleting {app}'s migration files")
    migrations_dir = os.path.join(settings.BASE_DIR, f'apps{os.path.sep}{app}{os.path.sep}migrations')

    if os.path.exists(migrations_dir):
        for the_file in os.listdir(migrations_dir):
            file_path = os.path.join(migrations_dir, the_file)
            try:
                if os.path.isfile(file_path):
                    os.unlink(file_path)
            except Exception as e:
                print(e)

            f = open(f"{os.path.join(migrations_dir, '__init__.py')}", "w")
            f.close()

    else:
        print('-' * 20, migrations_dir, 'does not exist')


class Command(BaseCommand):
    """
    Resets migrations and clears directories
    """
    help = 'reset migrations'

    def handle(self, *args, **options):
        set_of_apps = set()
        disregard = []

        # get all apps
        for model in django.apps.apps.get_models():
            if model._meta.app_label not in disregard:
                set_of_apps.add(model._meta.app_label)

        for app in set_of_apps:
            delete_migrations(app)

第 3 步

1. 从数据库中删除迁移(您可以使用下面的命令，它应该适用于任何使用 Postgres 的设置，但您必须根据需要更新连接字符串)

运行调用 python manage.py delete_migrations_from_db(如果你这样命名的话)

import os

import psycopg2
from django.conf import settings
from django.core.management import call_command
from django.core.management.base import BaseCommand
from django.db import connections


class Command(BaseCommand):
    help = 'Migrate on every database in settings.py'

    def handle(self, *args, **options):
        db_list = settings.DATABASES
        # del db_list['default']

        for db, _ in db_list.items():
            # we have the db name, now lets remove the migration tables in each

            try:
                host = os.environ['_HOST']
                user = os.environ['_USER']
                port = os.environ['_PORT']
                password = os.environ['_PASSWORD']

                conn_str = f"host={host} port={port} user={user} password={password}"

                conn = psycopg2.connect(conn_str)
                conn.autocommit = True
                with connections[db].cursor() as cursor:
                    delete_statement = 'DELETE from public.django_migrations'
                    cursor.execute(delete_statement)
                    print(f'Migration table cleared: {db}')
            except psycopg2.Error as ex:
                raise SystemExit(f'Error: {ex}')

        print('Done!')

第 4 步调用python manage.py makemigrations重新初始化初始迁移文件

第 5 步调用 python manage.py migrate --database=[YourDB] --fake 重新初始化初始迁移文件。 --fake arg 允许在数据库中恢复历史记录时不更改数据库结构(如果您想要一个简单的命令在所有数据库中运行此迁移命令，您可以使用类似代码如下)

使用 python manage.py migrate_all --fake 调用(取决于命名)

from django.conf import settings
from django.core.management import call_command
from django.core.management.base import BaseCommand


class Command(BaseCommand):
    help = 'Migrate on every database in settings.py'


    def add_arguments(self, parser):
        parser.add_argument(
            '--fake',
            action='store_true',
            help='fake migrations',
        )

    def handle(self, *args, **options):
        db_list = settings.DATABASES


        for db, _ in db_list.items():
            self.stdout.write('Migrating database {}'.format(db))

            if options['fake']:
                call_command('migrate', '--fake', database=db)
            else:
                # no fake, call regularly
                call_command('migrate', database=db)


        self.stdout.write('Done!')