fix: required columns on newly created tables retain migration safety checks

CHANGELOG.md

@@ -1,3 +1,9 @@
+## Unreleased
+
+Bug:
+- Do not report a required column added to a table unconditionally created earlier in the same migration.
+  Raw SQL, data migrations, and required columns added to existing tables remain linted.
+
 ## 6.0.0
 
 Feature:

DECISIONS.md

@@ -0,0 +1,12 @@
+# Architecture Decisions
+
+## Required columns on newly created tables
+
+**Decision.** The `NOT_NULL` SQL check accepts `ALTER TABLE ... ADD COLUMN ... NOT NULL`
+only when the same migration unconditionally created that exact table. `CREATE TABLE IF
+NOT EXISTS` does not prove an empty table. The migration remains
+fully linted, so `RunSQL`, `RunPython`, and operations on existing tables retain their
+checks.
+
+**Rejected.** Ignoring all initial migrations or naming them as exemptions. Either option
+skips unsafe operations that the linter must inspect.

src/django_migration_linter/sql_analyser/base.py

@@ -30,16 +30,41 @@ def update_migration_checks(
     return new_checks
 
 
+_IDENTIFIER = r'"[^"]+"|`[^`]+`|\'[^\']+\'|\[[^]]+\]|[A-Za-z_][A-Za-z0-9_$]*'
+_TABLE_NAME = rf"(?:{_IDENTIFIER})(?:\.(?:{_IDENTIFIER}))*"
+_CREATE_TABLE = re.compile(
+    rf"^CREATE TABLE (?!IF NOT EXISTS )(?P<table>{_TABLE_NAME})", re.IGNORECASE
+)
+_ADD_COLUMN = re.compile(
+    rf"^ALTER TABLE (?:ONLY )?(?P<table>{_TABLE_NAME}) ADD (?:COLUMN )?",
+    re.IGNORECASE,
+)
+
+
 def has_not_null_column(sql_statements: list[str], **kwargs) -> bool:
-    # TODO: improve to detect that the same column is concerned
+    """Detect a required column added to a table that already exists."""
     not_null_column = False
     has_default_value = False
+    created_tables = set()
 
     for sql in sql_statements:
-        if re.search("(?<!DROP )(?<!IS )NOT NULL", sql) and not (
+        created_table = _CREATE_TABLE.match(sql)
+        if created_table:
+            created_tables.add(created_table.group("table"))
+
+        adds_column = _ADD_COLUMN.match(sql)
+        adds_required_column = re.search("(?<!DROP )(?<!IS )NOT NULL", sql)
+        adds_to_new_table = (
+            adds_column is not None and adds_column.group("table") in created_tables
+        )
+        if (
+            adds_required_column
+            and not adds_to_new_table
+            and not (
                 sql.startswith("CREATE TABLE")
                 or sql.startswith("CREATE INDEX")
                 or sql.startswith("CREATE UNIQUE INDEX")
+            )
         ):
             not_null_column = True
         if re.search("DEFAULT (?!NULL).*NOT NULL", sql):

tests/unit/test_linter.py

@@ -5,9 +5,11 @@ import unittest
 from unittest.mock import patch
 
 from django.db import ProgrammingError
+from django.db import migrations
 from django.db.migrations import Migration
 
 from django_migration_linter import MigrationLinter
+from django_migration_linter.sql_analyser import get_sql_analyser_class
 
 
 class LinterFunctionsTestCase(unittest.TestCase):
@@ -77,7 +79,18 @@ class LinterFunctionsTestCase(unittest.TestCase):
         self.assertFalse(linter.should_ignore_migration("app_correct", "0003_bar"))
 
     def test_gather_all_migrations(self):
-        linter = MigrationLinter()
+        linter = MigrationLinter.__new__(MigrationLinter)
+        linter.exclude_migration_tests = []
+        linter.sql_analyser_class = get_sql_analyser_class("sqlite")
+        linter.include_apps = None
+        linter.exclude_apps = None
+        linter.ignore_name_contains = None
+        linter.include_name_contains = None
+        linter.ignore_name = []
+        linter.include_name = []
+        linter.only_applied_migrations = False
+        linter.only_unapplied_migrations = False
+        linter.ignore_initial_migrations = False
         migrations = linter._gather_all_migrations()
         self.assertGreater(len(list(migrations)), 1)
 
@@ -182,3 +195,39 @@ class LinterFunctionsTestCase(unittest.TestCase):
         self.assertFalse(
             linter.should_ignore_migration("app_correct", "0002_foo", is_initial=False)
         )
+
+    def test_initial_migration_still_lints_runsql_and_runpython(self):
+        def forward(apps, schema_editor):
+            return None
+
+        migration = Migration("0001_initial", "app_initial_operations")
+        migration.initial = True
+        migration.operations = [
+            migrations.RunSQL("DROP TABLE legacy_data;"),
+            migrations.RunPython(forward),
+        ]
+        linter = MigrationLinter.__new__(MigrationLinter)
+        linter.exclude_migration_tests = []
+        linter.sql_analyser_class = get_sql_analyser_class("sqlite")
+        linter.include_apps = None
+        linter.exclude_apps = None
+        linter.ignore_name_contains = None
+        linter.include_name_contains = None
+        linter.ignore_name = []
+        linter.include_name = []
+        linter.only_applied_migrations = False
+        linter.only_unapplied_migrations = False
+        linter.ignore_initial_migrations = False
+
+        errors, _, warnings = linter.analyse_data_migration(migration)
+
+        self.assertFalse(
+            linter.should_ignore_migration(
+                migration.app_label, migration.name, is_initial=migration.initial
+            )
+        )
+        self.assertEqual(["DROP_TABLE"], [issue.code for issue in errors])
+        self.assertEqual(
+            ["RUNSQL_REVERSIBLE", "RUNPYTHON_REVERSIBLE"],
+            [issue.code for issue in warnings],
+        )

tests/unit/test_sql_analyser.py

@@ -221,6 +221,34 @@ class SqliteAnalyserTestCase(SqlAnalyserTestCase):
 class PostgresqlAnalyserTestCase(SqlAnalyserTestCase):
     database_vendor = "postgresql"
 
+    def test_add_required_foreign_key_to_new_table(self):
+        sql = [
+            'CREATE TABLE "archive_capture" ("id" uuid NOT NULL PRIMARY KEY);',
+            'ALTER TABLE "archive_capture" ADD COLUMN "source_id" uuid NOT NULL REFERENCES "archive_source" ("id") DEFERRABLE INITIALLY DEFERRED;',
+        ]
+        self.assertValidSql(sql)
+
+    def test_add_required_foreign_key_to_existing_table(self):
+        sql = [
+            'CREATE TABLE "archive_capture" ("id" uuid NOT NULL PRIMARY KEY);',
+            'ALTER TABLE "other_app_capture" ADD COLUMN "source_id" uuid NOT NULL REFERENCES "archive_source" ("id") DEFERRABLE INITIALLY DEFERRED;',
+        ]
+        self.assertBackwardIncompatibleSql(sql, "NOT_NULL")
+
+    def test_add_required_foreign_key_after_conditional_create(self):
+        sql = [
+            'CREATE TABLE IF NOT EXISTS "legacy" ("id" uuid NOT NULL PRIMARY KEY);',
+            'ALTER TABLE "legacy" ADD COLUMN "source_id" uuid NOT NULL REFERENCES "archive_source" ("id") DEFERRABLE INITIALLY DEFERRED;',
+        ]
+        self.assertBackwardIncompatibleSql(sql, "NOT_NULL")
+
+    def test_add_required_foreign_key_to_new_qualified_table(self):
+        sql = [
+            'CREATE TABLE "archive"."capture" ("id" uuid NOT NULL PRIMARY KEY);',
+            'ALTER TABLE "archive"."capture" ADD COLUMN "source_id" uuid NOT NULL REFERENCES "archive"."source" ("id") DEFERRABLE INITIALLY DEFERRED;',
+        ]
+        self.assertValidSql(sql)
+
     def test_alter_column(self):
         sql = 'ALTER TABLE "app_alter_column_a" ALTER COLUMN "field" TYPE varchar(10) USING "field"::varchar(10);'
         self.assertBackwardIncompatibleSql(sql)