diff --git a/django/db/backends/mysql/introspection.py b/django/db/backends/mysql/introspection.py index a1ee51a6927b..ea1d0aa187e7 100644 --- a/django/db/backends/mysql/introspection.py +++ b/django/db/backends/mysql/introspection.py @@ -198,6 +198,7 @@ def get_relations(self, cursor, table_name): FROM information_schema.key_column_usage WHERE table_name = %s AND table_schema = DATABASE() + AND referenced_table_schema = DATABASE() AND referenced_table_name IS NOT NULL AND referenced_column_name IS NOT NULL """, @@ -257,6 +258,10 @@ def get_constraints(self, cursor, table_name): information_schema.table_constraints AS c WHERE kc.table_schema = DATABASE() AND + ( + kc.referenced_table_schema = DATABASE() OR + kc.referenced_table_schema IS NULL + ) AND c.table_schema = kc.table_schema AND c.constraint_name = kc.constraint_name AND c.constraint_type != 'CHECK' AND diff --git a/tests/backends/mysql/test_introspection.py b/tests/backends/mysql/test_introspection.py index c1247de2325c..a2e2938451a9 100644 --- a/tests/backends/mysql/test_introspection.py +++ b/tests/backends/mysql/test_introspection.py @@ -61,3 +61,63 @@ def test_get_storage_engine(self): cursor.execute(drop_sql) with other_connection.cursor() as cursor: cursor.execute(drop_sql) + + +@skipUnless(connection.vendor == "mysql", "MySQL specific SQL") +class TestCrossDatabaseRelations(TestCase): + databases = {"default", "other"} + + def test_omit_cross_database_relations(self): + default_connection = connections["default"] + other_connection = connections["other"] + main_table = "cross_schema_get_relations_main_table" + main_table_quoted = default_connection.ops.quote_name(main_table) + other_schema_quoted = other_connection.ops.quote_name( + other_connection.settings_dict["NAME"] + ) + rel_table = "cross_schema_get_relations_rel_table" + rel_table_quoted = other_connection.ops.quote_name(rel_table) + rel_column = "cross_schema_get_relations_rel_table_id" + rel_column_quoted = other_connection.ops.quote_name(rel_column) + try: + with other_connection.cursor() as other_cursor: + other_cursor.execute( + f""" + CREATE TABLE {rel_table_quoted} ( + id integer AUTO_INCREMENT, + PRIMARY KEY (id) + ) + """ + ) + with default_connection.cursor() as default_cursor: + # Create table in the default schema with a cross-database + # relation. + default_cursor.execute( + f""" + CREATE TABLE {main_table_quoted} ( + id integer AUTO_INCREMENT, + {rel_column_quoted} integer NOT NULL, + PRIMARY KEY (id), + FOREIGN KEY ({rel_column_quoted}) + REFERENCES {other_schema_quoted}.{rel_table_quoted}(id) + ) + """ + ) + relations = default_connection.introspection.get_relations( + default_cursor, main_table + ) + constraints = default_connection.introspection.get_constraints( + default_cursor, main_table + ) + self.assertEqual(len(relations), 0) + rel_column_fk_constraints = [ + spec + for name, spec in constraints.items() + if spec["columns"] == [rel_column] and spec["foreign_key"] is not None + ] + self.assertEqual(len(rel_column_fk_constraints), 0) + finally: + with default_connection.cursor() as default_cursor: + default_cursor.execute(f"DROP TABLE IF EXISTS {main_table_quoted}") + with other_connection.cursor() as other_cursor: + other_cursor.execute(f"DROP TABLE IF EXISTS {rel_table_quoted}")