[sqlite] Reflect DEFERRABLE and INITIALLY options for foreign keys

Fixes: #8903

diff --git a/lib/sqlalchemy/dialects/sqlite/base.py b/lib/sqlalchemy/dialects/sqlite/base.py
index 11554fcc0cd15d93a458867f6fca50360823e4a2..fb9a19c89ba1d146194a3cd5957d5994978d109b 100644 (file)
--- a/lib/sqlalchemy/dialects/sqlite/base.py
+++ b/lib/sqlalchemy/dialects/sqlite/base.py
@@ -2462,6 +2462,8 @@ class SQLiteDialect(default.DefaultDialect):
                 r'REFERENCES +(?:(?:"(.+?)")|([a-z0-9_]+)) *\((.+?)\) *'
                 r"((?:ON (?:DELETE|UPDATE) "
                 r"(?:SET NULL|SET DEFAULT|CASCADE|RESTRICT|NO ACTION) *)*)"
+                r"((?:NOT +)?DEFERRABLE)?"
+                r"(?: +INITIALLY +(DEFERRED|IMMEDIATE))?"
             )
             for match in re.finditer(FK_PATTERN, table_data, re.I):
                 (
@@ -2471,7 +2473,9 @@ class SQLiteDialect(default.DefaultDialect):
                     referred_name,
                     referred_columns,
                     onupdatedelete,
-                ) = match.group(1, 2, 3, 4, 5, 6)
+                    deferrable,
+                    initially,
+                ) = match.group(1, 2, 3, 4, 5, 6, 7, 8)
                 constrained_columns = list(
                     self._find_cols_in_sig(constrained_columns)
                 )
@@ -2493,6 +2497,12 @@ class SQLiteDialect(default.DefaultDialect):
                         onupdate = token[6:].strip()
                         if onupdate and onupdate != "NO ACTION":
                             options["onupdate"] = onupdate
+
+                if deferrable:
+                    options["deferrable"] = "NOT" not in deferrable.upper()
+                if initially:
+                    options["initially"] = initially.upper()
+
                 yield (
                     constraint_name,
                     constrained_columns,

diff --git a/test/dialect/test_sqlite.py b/test/dialect/test_sqlite.py
index d460ef64e05e26f52764d7c4467cc2f59033d51e..69c735aa57e085c2514381170e75f7c5b0d16794 100644 (file)
--- a/test/dialect/test_sqlite.py
+++ b/test/dialect/test_sqlite.py
@@ -1878,6 +1878,19 @@ class ConstraintReflectionTest(fixtures.TestBase):
                 "ON UPDATE NO ACTION)"
             )
 
+            conn.exec_driver_sql(
+                "CREATE TABLE deferrable_test (id INTEGER PRIMARY KEY, "
+                "c1 INTEGER, c2 INTEGER, c3 INTEGER, c4 INTEGER, "
+                "CONSTRAINT fk1 FOREIGN KEY (c1) REFERENCES a1(id) "
+                "DEFERRABLE,"
+                "CONSTRAINT fk2 FOREIGN KEY (c2) REFERENCES a1(id) "
+                "NOT DEFERRABLE,"
+                "CONSTRAINT fk3 FOREIGN KEY (c3) REFERENCES a2(id) "
+                "DEFERRABLE INITIALLY DEFERRED,"
+                "CONSTRAINT fk4 FOREIGN KEY (c4) REFERENCES a2(id) "
+                "NOT DEFERRABLE INITIALLY IMMEDIATE)"
+            )
+
             conn.exec_driver_sql(
                 "CREATE TABLE cp ("
                 "q INTEGER check (q > 1 AND q < 6),\n"
@@ -2248,6 +2261,47 @@ class ConstraintReflectionTest(fixtures.TestBase):
             ],
         )
 
+    def test_foreign_key_deferrable_initially(self):
+        inspector = inspect(testing.db)
+        fks = inspector.get_foreign_keys("deferrable_test")
+        eq_(
+            fks,
+            [
+                {
+                    "referred_table": "a1",
+                    "referred_columns": ["id"],
+                    "referred_schema": None,
+                    "name": "fk1",
+                    "constrained_columns": ["c1"],
+                    "options": {"deferrable": True},
+                },
+                {
+                    "referred_table": "a1",
+                    "referred_columns": ["id"],
+                    "referred_schema": None,
+                    "name": "fk2",
+                    "constrained_columns": ["c2"],
+                    "options": {"deferrable": False},
+                },
+                {
+                    "referred_table": "a2",
+                    "referred_columns": ["id"],
+                    "referred_schema": None,
+                    "name": "fk3",
+                    "constrained_columns": ["c3"],
+                    "options": {"deferrable": True, "initially": "DEFERRED"},
+                },
+                {
+                    "referred_table": "a2",
+                    "referred_columns": ["id"],
+                    "referred_schema": None,
+                    "name": "fk4",
+                    "constrained_columns": ["c4"],
+                    "options": {"deferrable": False, "initially": "IMMEDIATE"},
+                },
+            ],
+        )
+
     def test_foreign_key_options_unnamed_inline(self):
         with testing.db.begin() as conn:
             conn.exec_driver_sql(