Fixed #30054 -- Implemented cascaded flush on SQLite.

This is required to maintain foreign key integrity when using
TransactionTestCase.available_apps.

Refs #30033, #14204, #20483.
This commit is contained in:
Simon Charette 2018-12-22 17:47:48 -05:00 committed by Tim Graham
parent d5af14aa84
commit ce8b65ac5e
1 changed files with 21 additions and 6 deletions

View File

@ -159,6 +159,27 @@ class DatabaseOperations(BaseDatabaseOperations):
return -1
def sql_flush(self, style, tables, sequences, allow_cascade=False):
if tables and allow_cascade:
# Simulate TRUNCATE CASCADE by recursively collecting the tables
# referencing the tables to be flushed.
query = """
WITH tables AS (
%s
UNION
SELECT sqlite_master.name
FROM sqlite_master
JOIN tables ON (
sql REGEXP %%s || tables.name || %%s
)
) SELECT name FROM tables;
""" % ' UNION '.join("SELECT '%s' name" % table for table in tables)
params = (
r'(?i)\s+references\s+("|\')?',
r'("|\')?\s*\(',
)
with self.connection.cursor() as cursor:
results = cursor.execute(query, params)
tables = [row[0] for row in results.fetchall()]
sql = ['%s %s %s;' % (
style.SQL_KEYWORD('DELETE'),
style.SQL_KEYWORD('FROM'),
@ -168,12 +189,6 @@ class DatabaseOperations(BaseDatabaseOperations):
# sql_flush() implementations). Just return SQL at this point
return sql
def execute_sql_flush(self, using, sql_list):
# To prevent possible violation of foreign key constraints, deactivate
# constraints outside of the transaction created in super().
with self.connection.constraint_checks_disabled():
super().execute_sql_flush(using, sql_list)
def adapt_datetimefield_value(self, value):
if value is None:
return None