Checking Foreign Keys in MySQL

Question

Is there a way to check foreign key integrity in MySQL? For example; is it possible to go through a database in information_schema checking each table for a constraint violation?

Answer 1

I wrote some SQL to do this.

First, a listing of constraints, tables, columns, and child table information must be gathered:

SELECT DISTINCT KEY_COLUMN_USAGE.CONSTRAINT_NAME, KEY_COLUMN_USAGE.TABLE_NAME, KEY_COLUMN_USAGE.COLUMN_NAME, KEY_COLUMN_USAGE.REFERENCED_TABLE_NAME, KEY_COLUMN_USAGE.REFERENCED_COLUMN_NAME 
FROM INFORMATION_SCHEMA.TABLE_CONSTRAINTS 
INNER JOIN INFORMATION_SCHEMA.KEY_COLUMN_USAGE 
ON TABLE_CONSTRAINTS.CONSTRAINT_NAME=KEY_COLUMN_USAGE.CONSTRAINT_NAME 
WHERE TABLE_CONSTRAINTS.CONSTRAINT_TYPE="FOREIGN KEY" AND TABLE_CONSTRAINTS.CONSTRAINT_SCHEMA=<YOUR_DATABASE>;

Once this information is gathered you then loop through the returned data with the following SQL:

SELECT PARENT_TABLE.COLUMN AS CHILD_ID, CHILD_TABLE.CHILD_REFERENCED_COLUMN AS PARENT_ID FROM <YOUR_DATABASE>.<THE_TABLE> 
LEFT JOIN <YOUR_DATABASE>.CHILD_TABLE ON PARENT_TABLE.COLUMN=CHILD_TABLE.CHILD_REFERENCED_COLUMN 
WHERE CHILD_TABLE.CHILD_REFERENCED_COLUMN IS NULL;

Python code to do this is below:

self.dbcur.execute("SELECT DISTINCT KEY_COLUMN_USAGE.CONSTRAINT_NAME, KEY_COLUMN_USAGE.TABLE_NAME, KEY_COLUMN_USAGE.COLUMN_NAME, KEY_COLUMN_USAGE.REFERENCED_TABLE_NAME, KEY_COLUMN_USAGE.REFERENCED_COLUMN_NAME FROM INFORMATION_SCHEMA.TABLE_CONSTRAINTS INNER JOIN INFORMATION_SCHEMA.KEY_COLUMN_USAGE ON TABLE_CONSTRAINTS.CONSTRAINT_NAME=KEY_COLUMN_USAGE.CONSTRAINT_NAME WHERE TABLE_CONSTRAINTS.CONSTRAINT_TYPE=\"FOREIGN KEY\" AND TABLE_CONSTRAINTS.CONSTRAINT_SCHEMA=%s", (self.database,))
prep = self.dbcur.fetchall()

# 0 = CONSTRAINT_NAME
# 1 = TABLE_NAME
# 2 = COLUMN_NAME
# 3 = REFERENCED_TABLE_NAME
# 4 = REFERENCED_COLUMN_NAME
# 5 = DATABASE
for row in prep:
    query = "SELECT {1}.{2} AS CHILD_ID, {3}.{4} AS PARENT_ID FROM {5}.{1} LEFT JOIN {5}.{3} ON {1}.{2}={3}.{4} WHERE {3}.{4} IS NULL".format(row[0], row[1], row[2], row[3], row[4], self.database)
    self.dbcur.execute(query)
    results = self.dbcur.fetchall()

    if len(results) > 0:
        for baddata in results:
            bad_data = ({"CONSTRAINT_NAME": row[0], "TABLE_NAME": row[1], "COLUMN_NAME": row[2], "REFERENCED_TABLE_NAME": row[3], "REFERENCED_COLUMN_NAME": row[4], "CHILD_ID": baddata[0], "PARENT_ID": baddata[1], "DATABASE": self.database})
            self.badforeignkeys.append(bad_data)

return(self.badforeignkeys)