Silent Django Migrations

django_silent_migrations.py

from django.db.migrations.operations.base import Operation


class SilentOperationWrapper(Operation):
    """
    A wrapper for Django migration operations that handles errors silently.

    This wrapper executes the wrapped operation and catches any exceptions.
    If an exception occurs, it can either fail silently or mark the operation
    as "faked" in the migration history without actually performing the operation.

    Example usage:
        operations = [
            SilentOperationWrapper(
                operations.CreateModel(...),
                fake_on_error=True,
                log_errors=True
            )
        ]
    """

    def __init__(self, operation, fake_on_error=True, log_errors=True):
        """
        Initialize the wrapper.

        Args:
            operation: The Django migration operation to wrap
            fake_on_error: Whether to fake the operation on error (True) or
                           just suppress the error (False)
            log_errors: Whether to log errors when they occur
        """
        self.operation = operation
        self.fake_on_error = fake_on_error
        self.log_errors = log_errors

    def deconstruct(self):
        kwargs = {
            "operation": self.operation,
            "fake_on_error": self.fake_on_error,
            "log_errors": self.log_errors,
        }
        return (self.__class__.__qualname__, [], kwargs)

    def state_forwards(self, app_label, state):
        """
        Always update the state as if the operation succeeded.
        This is needed for subsequent operations to work correctly.
        """
        try:
            self.operation.state_forwards(app_label, state)
        except Exception as e:
            if self.log_errors:
                import logging

                logger = logging.getLogger("django.db.migrations")
                logger.warning(
                    f"Error in state_forwards for {self.operation.__class__.__name__}: {e}",
                    exc_info=True,
                )

    def database_forwards(self, app_label, schema_editor, from_state, to_state):
        """
        Execute the wrapped operation's database_forwards method,
        handling errors based on the configuration.
        """
        try:
            self.operation.database_forwards(
                app_label, schema_editor, from_state, to_state
            )
        except Exception as e:
            if self.log_errors:
                import logging

                logger = logging.getLogger("django.db.migrations")
                logger.warning(
                    f"Error in database_forwards for {self.operation.__class__.__name__}: {e}",
                    exc_info=True,
                )
            # If we should fake on error, record the migration but don't perform it
            if self.fake_on_error and hasattr(schema_editor, "record_migration"):
                schema_editor.record_migration(
                    self.operation.__class__.__name__, app_label
                )

    def database_backwards(self, app_label, schema_editor, from_state, to_state):
        """
        Execute the wrapped operation's database_backwards method,
        handling errors based on the configuration.
        """
        try:
            self.operation.database_backwards(
                app_label, schema_editor, from_state, to_state
            )
        except Exception as e:
            if self.log_errors:
                import logging

                logger = logging.getLogger("django.db.migrations")
                logger.warning(
                    f"Error in database_backwards for {self.operation.__class__.__name__}: {e}",
                    exc_info=True,
                )
            # If we should fake on error, record the migration but don't perform it
            if self.fake_on_error and hasattr(schema_editor, "record_migration"):
                schema_editor.record_migration(
                    self.operation.__class__.__name__, app_label
                )

    def describe(self):
        """Return a description of the wrapped operation with additional info."""
        return f"Silent wrapper for: {self.operation.describe()}"

    # Forward properties and methods to the wrapped operation
    @property
    def reversible(self):
        return getattr(self.operation, "reversible", True)

    @property
    def migration_name_fragment(self):
        if hasattr(self.operation, "migration_name_fragment"):
            return f"silent_{self.operation.migration_name_fragment}"
        return "silent_operation"

    def references_model(self, name, app_label=None):
        if hasattr(self.operation, "references_model"):
            return self.operation.references_model(name, app_label)
        return False

    def references_field(self, model_name, name, app_label=None):
        if hasattr(self.operation, "references_field"):
            return self.operation.references_field(model_name, name, app_label)
        return False

    def allow_migrate_model(self, connection_alias, model):
        if hasattr(self.operation, "allow_migrate_model"):
            return self.operation.allow_migrate_model(connection_alias, model)
        return True