Resolving MySQL 1215 errors in a declarative MySQL migration system

Question

Github here: https://github.com/cmancone/mygrations

I'm building a migration system for MySQL that operates in a very different way than the usual. Rather than recording changes in the database, the database structure is managed by simply having one CREATE TABLE command for each MySQL table in any number of .sql files. The migration system reads these files to determine the desired database structure, reads the database to determine the current database structure, and then calculates the SQL commands needed to migrate one to the other.

One big advantage for doing it this way is that it can be intelligent about foreign key constraints, and in particular can resolve those obnoxious 1215 errors. The system obviously records foreign key constraints while processing, and can check a foreign key constraint against a database to determine whether or not the foreign key constraint is valid. If it isn't, it can generate a specific error message rather than the generic 1215 error that MySQL returns for any issues.

I'm getting very close to a first "alpha" stage and would like some general feedback on a few key parts of the process. One of those is the part that actually checks for any 1215 errors and returns a useful error message. It lives in my database class. This class is not directly connected to an actual database, but instead just represents an in-memory store of a given database's structure. Other classes are responsible for using it to actually do things (build migration plans, update database, etc), so there isn't a ton of functionality in here. Some notes/comments/desires:

1. Docstrings follow the sphinx standard
2. Code has been converted to PEP8 as much as possible for this review (I use a slight variation on PEP8)
3. The database class keeps most of its data in tables which is a Dict of table objects
4. The table objects have OrderedDicts for columns, indexes, and constraints which you will see referenced in the code.
5. Most of the behavior of interest lies in the find_1215_errors method, which returns a simple string as an error message if a problem is found, and False otherwise
6. Bonus points if you find a MySQL 1215 error condition that I have missed. To the best of my knowledge, the only one that I'm missing is if you attempt to add a constraint to a table which has data that is not consistent with the foreign key constraint. However, this kind of foreign key error has to be checked in a completely different way, so is not applicable here.

The database class:

class database(object):
    _errors = None
    _warnings = None
    _tables = None
    _rows = None

    def __init__(self):
        self._warnings = []
        self._errors = []
        self._tables = {}
        self._rows = []

    @property
    def tables(self):
        """ Public getter.  Returns a dict of table definitions, by table name

        :returns: A dict of table definitions, by table name
        :rtype: dict
        """
        return {} if self._tables is None else self._tables

    @property
    def errors( self ):
        """ Public getter.  Returns a list of parsing errors

        :returns: A list of parsing errors
        :rtype: list
        """
        return [] if self._errors is None else self._errors

    @property
    def warnings( self ):
        """ Public getter.  Returns a list of parsing/table warnings

        :returns: A list of parsing/table warnings
        :rtype: list
        """
        return [] if self._warnings is None else self._warnings

    def unfulfilled_fks( self, table ):
        """ Returns a dictionary with information about all constraints in the table which are not fulfilled by this database

        If all foreign keys are fulfilled by the database structure then an empty dict is returned

        The returned dictionary contains a key with the name of every foreign key that cannot
        be fulfilled.  The value in the dictionary will be another dictionary containing
        'error' (an error message stating exactly what the problem is) and 'foreign_key'
        (the actual foreign key definition that cannot be fulfilled)

        :param table: The table to check
        :type table: mygrations.formats.mysql.definitions.table
        :return: Dictionary with information on all foreign keys that cannot be fulfilled
        :rtype: dict
        """
        if not table.constraints:
            return {}

        unfulfilled = {}
        for (constraint_name, constraint) in table.constraints.items():
            error = self.find_1215_errors(table, constraint)
            if error:
                unfulfilled[constraint_name] = {"error": error, "foreign_key": constraint}

        return unfulfilled

    def find_1215_errors( self, table, constraint ):
        """ Returns False or a string describing a 1215 error message found for the given table and constraint

        :param table: The table being checked
        :param constraint: The constraint to check the table against
        :type table: mygrations.formats.mysql.definitions.table
        :type constraint: mygrations.formats.mysql.definitions.constraint
        :rtype: string|False
        """
        if constraint.foreign_table not in self.tables:
            return "MySQL 1215 error for foreign key `%s`: `%s`.`%s` references `%s`.`%s`, but table `%s` does not exist" % (
                constraint.name,
                table.name,
                constraint.column,
                constraint.foreign_table,
                constraint.foreign_column,
                constraint.foreign_table)

        foreign_table = self.tables[constraint.foreign_table]
        if constraint.foreign_column not in foreign_table.columns:
            return "MySQL 1215 error for foreign key `%s`: `%s`.`%s` references `%s`.`%s`, but column `%s`.`%s` does not exist" % (
                constraint.name,
                table.name,
                constraint.column,
                constraint.foreign_table,
                constraint.foreign_column,
                constraint.foreign_table,
                constraint.foreign_column)

        # the column exists but we may still have a 1215 error.  That can happen in a few ways
        table_column = table.columns[constraint.column]
        foreign_column = foreign_table.columns[constraint.foreign_column]

        # we have a few attributes that must must match exactly and have easy-to-produce errors
        for attr in [ 'column_type', 'length', 'character_set', 'collate' ]:
            table_value = getattr( table_column, attr )
            foreign_value = getattr( foreign_column, attr )
            if table_value != foreign_value:
                return "MySQL 1215 error for foreign key `%s`: %s mismatch. `%s`.`%s` is '%s' but `%s`.`%s` is '%s'" % (
                    constraint.name,
                    attr.replace( '_', ' ' ),
                    table.name,
                    constraint.column,
                    table_value,
                    foreign_table.name,
                    foreign_column.name,
                    foreign_value)

        # unsigned are separate because they get a slightly different message
        if table_column.unsigned and not foreign_column.unsigned:
            return "MySQL 1215 error for foreign key `%s`: unsigned mistmatch. `%s`.`%s` is unsigned but `%s`.`%s` is not" % (
                constraint.name,
                table.name,
                table_column.name,
                foreign_table.name,
                foreign_column.name)

        if not table_column.unsigned and foreign_column.unsigned:
            return "MySQL 1215 error for foreign key `%s`: unsigned mistmatch. `%s`.`%s` is unsigned but `%s`.`%s` is not" % (
                constraint.name,
                foreign_table.name,
                foreign_column.name,
                table.name,
                table_column.name)

        # if the constraint has a SET NULL but the column cannot be null, then 1215
        if ( constraint.on_delete == 'SET NULL' or constraint.on_update == 'SET NULL' ) and not table_column.null:
            message_parts = []
            if constraint.on_delete == 'SET NULL':
                message_parts.append( 'ON DELETE' )
            if constraint.on_update == 'SET NULL':
                message_parts.append( 'ON UPDATE' )
            return "MySQL 1215 error for foreign key `%s`: invalid SET NULL. `%s`.`%s` is not allowed to be null but the foreign key attempts to set the value to null %s" % (
                constraint.name,
                table.name,
                table_column.name,
                ' and '.join(message_parts))

        # if the column the constraint is on doesn't have an index, then 1215
        index_found = False
        for index in foreign_table.indexes.values():
            if index.columns[0] != foreign_column.name:
                continue
            index_found = True
            break

        if not index_found:
            return "MySQL 1215 error for foreign key `%s`: missing index. `%s`.`%s` references `%s`.`%s` but `%s`.`%s` does not have an index and therefore cannot be used in a foreign key constraint" % (
                constraint.name,
                table.name,
                table_column.name,
                foreign_table.name,
                foreign_column.name,
                foreign_table.name,
                foreign_column.name)
        return False

    def add_table(self, table):
        """ Adds a table to the database

        :param table: The table to add
        :type table: mygrations.formats.mysql.definitions.table
        """
        if table.name in self._tables:
            raise ValueError('Cannot add table %s to database because it already exists' % table.name)
        self._tables[table.name] = table

    def remove_table(self, table):
        """ Removes a table from the database

        :param table: The table to remove
        :type table: mygrations.formats.mysql.definitions.table
        """
        if not table.name in self._tables:
            raise ValueError('Cannot remove table %s from database because it does not exist' % table.name)
        self._tables.pop(table.name, None)

    def apply_operation(self, table_name, operation):
        """ Applies an operation to the database

        :param table_name: The table that the operation is being applied to
        :param operation: The operation to apply
        :type table: string|mygrations.formats.mysql.definitions.table
        :type operation: mygrations.formats.mysql.mygration.operations.*
        """
        if type( table_name ) != str:
            table_name = table_name.name
        if not table_name in self._tables:
            raise ValueError('Cannot apply operation to table %s because that table does not exist' % table_name)
        self._tables[table_name].apply_operation(operation)

One specific question Right now all the foreign key constraint checks happen in the database object, with the database object checking the columns of its child tables. I also considered having the database simply confirm that the table being reference exists, and then deferring the rest of the checking to the table. I would then add a find_1215_errors public method to the table class, and basically all the logic would just copy over to there. I could similarly add some of the column-specific checks to the underlying column objects, which the table would defer to. This certainly makes much more sense from an encapsulation perspective, but I like having all the checks in one place. I don't foresee a need to check foreign keys at the column-level only, so I don't think having the checks at the database-only will ever cause trouble. I'm happy to hear an outsider perspective on the issue though.

In case it is helpful the code for the table class can be found here:

https://github.com/cmancone/mygrations/blob/mysql/mygrations/formats/mysql/definitions/table.py

And column here:

https://github.com/cmancone/mygrations/blob/mysql/mygrations/formats/mysql/definitions/column.py

I think the database class is understandable without seeing these, so I'm leaving them out for the sake of space.

Answer 1

Thank you for docstrings! You have clearly devoted effort to them - consider pointing to an URL of sphinx output.

In tables, errors, and

@property
def warnings( self ):
    return [] if self._warnings is None else self._warnings

I don't understand why we don't simply return self._warnings, given that __init__ assigned []. Also, PEP-8, no blanks around self. Running flake8 would tidy that up.

I think your recent edit added these four assignments:

class database(object):
    _errors = None
    _warnings = None
    _tables = None
    _rows = None

It would be helpful to add a comment mentioning the __new__ vs. __init__ distinction you're going for.

    ... The value in the dictionary will be another dictionary containing
    'error' (an error message stating exactly what the problem is) and 'foreign_key'
    (the actual foreign key definition that cannot be fulfilled)

That is a perfectly nice public API, using the flexibility of a dict. A more natural API might return a pair (a tuple) instead of a 2nd dict.

    """ Returns False or a string describing a 1215 error message found for the given table and constraint

This also is a clear API, but returning None would be more pythonic. Elsewhere you have several trinary functions that very nicely return True / False / None - that makes perfect sense.

Pair of copy-n-paste typos: "unsigned mistmatch"

The pattern of returning a formatted error string is good, and the one-parameter-per-line formatting is easy to read. There is enough copy-n-paste duplication there to motivate extracting a helper function or two, for some error string fragments.

    # if the column the constraint is on doesn't have an index, then 1215
    index_found = False
    for index in foreign_table.indexes.values():
        if index.columns[0] != foreign_column.name:
            continue
        index_found = True
        break

This seems verbose. Consider building a set() and testing set membership.

This tool seems intended for repeated quick runs over a schema, as it may bail out early upon detecting first error, rather than accumulating a batch of all errors and reporting them together.

In your One Specific Question you wondered whether a certain design decision was the right one. My vote is yes, you made a good choice. It causes the method to become slightly on the long side, but that's fine here. It has clear repetitive logic, and it is good to see all the tests together. At some point a maintainer will likely have to add Yet Another Test for MariaDB 15, and it will be perfectly clear where to start reading, and where to add that Nth test. Good job!