Allow db only remove foreign key to work as database only

CHANGELOG.md

@@ -8,7 +8,11 @@ Changelog](https://keepachangelog.com/en/1.1.0/), and this project adheres to
 
 ## [Unreleased]
 
-_No notable unreleased changes_
+### Fixed
+
+- Fixed a bug where `SaferRemoveFieldForeignKey` relied on the Foreign Key also
+  existing in Django state, even when being performed as a database only operation.
+  The name and model are already provided as part of the operation.
 
 ## [0.1.23] - 2025-11-18
 

src/django_pg_migration_tools/operations.py

@@ -5,6 +5,7 @@
 from typing import Any, cast, overload
 
 from django.contrib.postgres import operations as psql_operations
+from django.core import exceptions
 from django.db import migrations, models
 from django.db.backends import utils as django_backends_utils
 from django.db.backends.base import schema as base_schema
@@ -1154,11 +1155,11 @@ def __init__(
         model: type[models.Model],
         model_name: str,
         column_name: str,
-        field: models.ForeignKey[models.Model],
+        field: models.ForeignKey[models.Model] | None,
         unique: bool,
         skip_null_check: bool = False,
     ) -> None:
-        if not field.null and not skip_null_check:
+        if field is not None and (not field.null and not skip_null_check):
             # Validate at initialisation, rather than wasting time later.
             raise ValueError("Can't safely create a FK field with null=False")
 
@@ -1242,6 +1243,7 @@ def add_fk_field(self) -> None:
             # as they would add extra introspection queries unnecessarily.
             self._maybe_create_unique_constraint()
 
+        assert self.field is not None
         assert hasattr(self.field, "db_index")
         if (
             self.field.db_index
@@ -1289,6 +1291,7 @@ def _column_exists(self, collect_default: bool = False) -> bool:
         )
 
     def _get_remote_model(self) -> models.Model:
+        assert self.field is not None
         if isinstance(self.field.remote_field.model, str):
             app_name, model_name = self.field.remote_field.model.split(".")  # type: ignore[unreachable]
 
@@ -1307,6 +1310,7 @@ def _get_remote_pk_field(self) -> models.Field[Any, Any]:
         return pk_field
 
     def _get_remote_to_field(self) -> models.Field[Any, Any]:
+        assert self.field is not None
         to_field = self.field.to_fields[0]
         remote_model = self._get_remote_model()
 
@@ -1318,6 +1322,7 @@ def _get_remote_to_field(self) -> models.Field[Any, Any]:
 
     def _get_target_field(self) -> models.Field[Any, Any]:
         # If to_field is specified, we don't want to default to using the pk.
+        assert self.field is not None
         if self.field.to_fields and self.field.to_fields[0]:
             target_field = self._get_remote_to_field()
         else:
@@ -1356,6 +1361,7 @@ def _maybe_create_index(self) -> None:
             # be used as indexes by Postgres.
             return
 
+        assert self.field is not None
         assert hasattr(self.field, "db_index")
         if self.field.db_index:
             SafeIndexOperationManager().safer_create_index(
@@ -1492,9 +1498,12 @@ def database_forwards(
         from_state: migrations.state.ProjectState,
         to_state: migrations.state.ProjectState,
     ) -> None:
-        field = from_state.apps.get_model(app_label, self.model_name)._meta.get_field(
-            self.name
-        )
+        try:
+            field = from_state.apps.get_model(
+                app_label, self.model_name
+            )._meta.get_field(self.name)
+        except exceptions.FieldDoesNotExist:
+            field = None
         ForeignKeyManager(
             app_label,
             schema_editor,

tests/django_pg_migration_tools/test_operations.py

@@ -2588,6 +2588,160 @@ def test_operation(self):
                 AND convalidated IS TRUE;
         """)
 
+    @pytest.mark.django_db(transaction=True)
+    def test_operation_when_already_removed_from_state(self):
+        with connection.cursor() as cursor:
+            # Set the lock_timeout to check it has been returned to
+            # its original value once the fk index creation is completed by
+            # the reverse operation.
+            cursor.execute(_SET_LOCK_TIMEOUT)
+
+        project_state = ProjectState()
+        project_state.add_model(ModelState.from_model(IntModel))
+        project_state.add_model(ModelState.from_model(ModelWithForeignKey))
+        new_state = project_state.clone()
+        operation = operations.SaferRemoveFieldForeignKey(
+            model_name="modelwithforeignkey",
+            name="fk",
+        )
+
+        assert operation.describe() == (
+            "Remove field fk from modelwithforeignkey. Note: Using "
+            "django_pg_migration_tools SaferRemoveFieldForeignKey operation."
+        )
+
+        operation.state_forwards(self.app_label, new_state)
+
+        # Do database only operation - has already been removed from state
+        newer_state = new_state.clone()
+        with connection.schema_editor(atomic=False, collect_sql=False) as editor:
+            with utils.CaptureQueriesContext(connection) as queries:
+                operation.database_forwards(
+                    self.app_label, editor, from_state=new_state, to_state=newer_state
+                )
+
+        assert len(queries) == 2
+
+        assert queries[0]["sql"] == dedent(
+            """
+            SELECT 1
+            FROM pg_catalog.pg_attribute
+            WHERE
+                attrelid = 'example_app_modelwithforeignkey'::regclass
+                AND attname = 'fk_id';
+        """
+        )
+        assert queries[1]["sql"] == dedent(
+            """
+            ALTER TABLE "example_app_modelwithforeignkey"
+            DROP COLUMN "fk_id";
+        """
+        )
+
+        with connection.schema_editor(atomic=False, collect_sql=False) as editor:
+            with utils.CaptureQueriesContext(connection) as reverse_queries:
+                operation.database_backwards(
+                    self.app_label, editor, from_state=new_state, to_state=project_state
+                )
+
+        assert len(reverse_queries) == 9
+
+        assert reverse_queries[0]["sql"] == dedent(
+            """
+            SELECT 1
+            FROM pg_catalog.pg_attribute
+            WHERE
+                attrelid = 'example_app_modelwithforeignkey'::regclass
+                AND attname = 'fk_id';
+        """
+        )
+        assert reverse_queries[1]["sql"] == dedent(
+            """
+            ALTER TABLE "example_app_modelwithforeignkey"
+            ADD COLUMN IF NOT EXISTS "fk_id"
+            integer NULL;
+        """
+        )
+        assert reverse_queries[2]["sql"] == "SHOW lock_timeout;"
+        assert reverse_queries[3]["sql"] == "SET lock_timeout = '0';"
+        assert reverse_queries[4]["sql"] == dedent(
+            """
+            SELECT relname
+            FROM pg_class, pg_index
+            WHERE (
+                pg_index.indisvalid = false
+                AND pg_index.indexrelid = pg_class.oid
+                AND relname = 'modelwithforeignkey_fk_id_idx'
+            );
+            """
+        )
+        assert (
+            reverse_queries[5]["sql"]
+            == 'CREATE INDEX CONCURRENTLY IF NOT EXISTS "modelwithforeignkey_fk_id_idx" ON "example_app_modelwithforeignkey" ("fk_id");'
+        )
+        assert reverse_queries[6]["sql"] == "SET lock_timeout = '1s';"
+        assert reverse_queries[7]["sql"] == dedent(
+            """
+            ALTER TABLE "example_app_modelwithforeignkey"
+            ADD CONSTRAINT "example_app_modelwithforeignkey_fk_id_fk" FOREIGN KEY ("fk_id")
+            REFERENCES "example_app_intmodel" ("id")
+            DEFERRABLE INITIALLY DEFERRED
+            NOT VALID;
+        """
+        )
+        assert reverse_queries[8]["sql"] == dedent(
+            """
+            ALTER TABLE "example_app_modelwithforeignkey"
+            VALIDATE CONSTRAINT "example_app_modelwithforeignkey_fk_id_fk";
+        """
+        )
+
+        # Reversing again does nothing apart from checking that the FK is
+        # already there and the index/constraint are all good to go.
+        # This proves the OP is idempotent.
+        with connection.schema_editor(atomic=False, collect_sql=False) as editor:
+            with utils.CaptureQueriesContext(connection) as second_reverse_queries:
+                operation.database_backwards(
+                    self.app_label, editor, from_state=new_state, to_state=project_state
+                )
+        assert len(second_reverse_queries) == 4
+        assert second_reverse_queries[0]["sql"] == dedent(
+            """
+            SELECT 1
+            FROM pg_catalog.pg_attribute
+            WHERE
+                attrelid = 'example_app_modelwithforeignkey'::regclass
+                AND attname = 'fk_id';
+        """
+        )
+        assert second_reverse_queries[1]["sql"] == dedent(
+            """
+            SELECT 1
+            FROM pg_class, pg_index
+            WHERE (
+                pg_index.indisvalid = true
+                AND pg_index.indexrelid = pg_class.oid
+                AND relname = 'modelwithforeignkey_fk_id_idx'
+            );
+        """
+        )
+        assert second_reverse_queries[2]["sql"] == dedent(
+            """
+            SELECT conname
+            FROM pg_catalog.pg_constraint
+            WHERE conname = 'example_app_modelwithforeignkey_fk_id_fk';
+        """
+        )
+        assert second_reverse_queries[3]["sql"] == dedent(
+            """
+            SELECT 1
+            FROM pg_catalog.pg_constraint
+            WHERE
+                conname = 'example_app_modelwithforeignkey_fk_id_fk'
+                AND convalidated IS TRUE;
+        """
+        )
+
     @pytest.mark.django_db(transaction=True)
     def test_when_column_not_null(self):
         with connection.cursor() as cursor: