Add Postgresql warning when using CREATE INDEX without CONCURRENTLY keyword

django_migration_linter/sql_analyser/analyser.py

@@ -29,4 +29,4 @@ def analyse_sql_statements(
 ):
     sql_analyser = get_sql_analyser(database_vendor, exclude_migration_tests)
     sql_analyser.analyse(sql_statements)
-    return sql_analyser.errors, sql_analyser.ignored
+    return sql_analyser.errors, sql_analyser.ignored, sql_analyser.warnings

django_migration_linter/sql_analyser/postgresql.py

@@ -1,5 +1,16 @@
+import re
+
 from .base import BaseAnalyser
 
 
 class PostgresqlAnalyser(BaseAnalyser):
-    migration_tests = []
+    migration_tests = [
+        {
+            "code": "CREATE_INDEX",
+            "fn": lambda sql, **kw: re.search("CREATE (UNIQUE )?INDEX", sql)
+            and not re.search("INDEX CONCURRENTLY", sql),
+            "msg": "CREATE INDEX locks table",
+            "mode": "one_liner",
+            "type": "warning",
+        },
+    ]

tests/unit/test_sql_analyser.py

@@ -14,13 +14,17 @@ class SqlAnalyserTestCase(unittest.TestCase):
         )
 
     def assertValidSql(self, sql):
-        errors, _ = self.analyse_sql(sql)
+        errors, _, _ = self.analyse_sql(sql)
         self.assertEqual(0, len(errors), "Found errors in sql: {}".format(errors))
 
     def assertBackwardIncompatibleSql(self, sql):
-        errors, _ = self.analyse_sql(sql)
+        errors, _, _ = self.analyse_sql(sql)
         self.assertNotEqual(0, len(errors), "Found no errors in sql")
 
+    def assertWarningSql(self, sql):
+        _, _, warnings = self.analyse_sql(sql)
+        self.assertNotEqual(0, len(warnings), "Found no warnings in sql")
+
 
 class MySqlAnalyserTestCase(SqlAnalyserTestCase):
     database_vendor = "mysql"
@@ -203,3 +207,15 @@ class PostgresqlAnalyserTestCase(SqlAnalyserTestCase):
             'ALTER TABLE "app_drop_default_a" ALTER COLUMN "col" SET DEFAULT `\'empty\';',
         ]
         self.assertValidSql(sql)
+
+    def test_create_index_non_concurrently(self):
+        sql = "CREATE INDEX ON films ((lower(titre)));"
+        self.assertWarningSql(sql)
+        sql = "CREATE UNIQUE INDEX title_idx ON films (title);"
+        self.assertWarningSql(sql)
+
+    def test_create_index_concurrently(self):
+        sql = "CREATE INDEX CONCURRENTLY ON films ((lower(titre)));"
+        self.assertValidSql(sql)
+        sql = "CREATE UNIQUE INDEX CONCURRENTLY title_idx ON films (title);"
+        self.assertValidSql(sql)