Примените миграцию django, если ограничение еще не существует

Я хочу добавить ограничение в мою модель django. Я хочу, чтобы эта миграция ничего не делала, если ограничение уже было добавлено (вручную в sql, не django). Я знаю, как проверить, что ограничение уже применено, благодаря этому вопросу.

Однако, как я могу пропустить операцию "add constraint" в моем файле миграции, основываясь на этом условии? Нужно ли мне создавать операцию RunPython или я могу полагаться на объекты миграции django? Я использую postgres.

EDIT Я добавил ограничение ранее непосредственно в sql, чтобы не делать этого во время развертывания, по причине производительности. Кроме того, это ограничение применяется не во всех средах, поэтому я не могу предположить, что оно уже есть.

Посмотрите на "фальшивые" миграции. Позвольте django создать миграцию, как он бы это сделал, однако вы можете выполнить следующее действие, чтобы сказать Django "Я запустил миграцию", когда на самом деле вы этого не сделали, и он не будет предлагать эту миграцию снова:

python manage.py migrate --fake my_module

https://docs.djangoproject.com/en/4.0/ref/django-admin/#cmdoption-migrate-fake

Для других сред, к которым не применяется ограничение, просто пропустите вариант с подделкой

Мне удалось реализовать дочерний класс операции миграции AddConstraint. Сначала он проверяет, существует ли уже ограничение.

import warnings

from django.db import migrations


class SkipIfExistAddConstraint(migrations.AddConstraint):
    def database_forwards(self, app_label, schema_editor, from_state, to_state):
        db_table = to_state.apps.get_model(app_label, self.model_name)._meta.db_table
        if self.constraint_exists(schema_editor, db_table):
            warnings.warn(
                f"Constraint {self.constraint.name} of table {db_table} already exists. "
                "Skipping..."
            )
            return
        return super().database_forwards(app_label, schema_editor, from_state, to_state)

    def constraint_exists(self, schema_editor, db_table):
        with schema_editor.connection.cursor() as cursor:
            cursor.execute(
                (
                    "select 1 "
                    "from information_schema.constraint_column_usage "
                    f"where table_name = '{db_table}' "
                    f"and constraint_name = '{self.constraint.name}'"
                )
            )
            return cursor.fetchone() is not None
Вернуться на верх