diff --git a/django_migration_linter/cache.py b/django_migration_linter/cache.py index 0ddd08c..61d48a6 100644 --- a/django_migration_linter/cache.py +++ b/django_migration_linter/cache.py @@ -23,8 +23,7 @@ class Cache(dict): def __init__(self, django_folder, cache_path): self.django_folder = django_folder project_name = split_path(django_folder)[-2] - self.filename = \ - os.path.join(cache_path, '{0}.pickle'.format(project_name)) + self.filename = os.path.join(cache_path, "{0}.pickle".format(project_name)) if not os.path.exists(os.path.dirname(self.filename)): os.makedirs(os.path.dirname(self.filename)) @@ -33,14 +32,14 @@ class Cache(dict): def load(self): try: - with open(self.filename, 'rb') as f: + with open(self.filename, "rb") as f: tmp_dict = pickle.load(f) self.update(tmp_dict) except IOError: pass def save(self): - with open(self.filename, 'wb') as f: + with open(self.filename, "wb") as f: pickle.dump(self, f, protocol=2) def md5(self, app_name, migration): diff --git a/django_migration_linter/constants.py b/django_migration_linter/constants.py index 1bdfc9b..a95e8dc 100644 --- a/django_migration_linter/constants.py +++ b/django_migration_linter/constants.py @@ -14,10 +14,8 @@ from appdirs import user_cache_dir -__version__ = '0.0.7' +__version__ = "0.0.7" -MIGRATION_FOLDER_NAME = 'migrations' +MIGRATION_FOLDER_NAME = "migrations" -DEFAULT_CACHE_PATH = user_cache_dir( - "django-migration-linter", version=__version__ -) +DEFAULT_CACHE_PATH = user_cache_dir("django-migration-linter", version=__version__) diff --git a/django_migration_linter/migration_linter.py b/django_migration_linter/migration_linter.py index e3d1fd3..d88f504 100644 --- a/django_migration_linter/migration_linter.py +++ b/django_migration_linter/migration_linter.py @@ -20,8 +20,7 @@ from subprocess import Popen, PIPE import sys from django_migration_linter.cache import Cache -from django_migration_linter.constants import DEFAULT_CACHE_PATH, \ - MIGRATION_FOLDER_NAME +from django_migration_linter.constants import DEFAULT_CACHE_PATH, MIGRATION_FOLDER_NAME from django_migration_linter.utils import split_migration_path from . import utils from .sql_analyser import analyse_sql_statements @@ -30,30 +29,36 @@ logger = logging.getLogger(__name__) class MigrationLinter(object): - def __init__(self, project_path, **kwargs): # Verify correctness if not utils.is_directory(project_path): raise ValueError( - 'The given path {0} does not seem to be a directory.'.format( - project_path)) + "The given path {0} does not seem to be a directory.".format( + project_path + ) + ) if not utils.is_django_project(project_path): raise ValueError( - ('The given path {0} does not ' - 'seem to be a django project.').format(project_path)) + ("The given path {0} does not " "seem to be a django project.").format( + project_path + ) + ) # Store parameters and options self.django_path = project_path - self.ignore_name_contains = kwargs.get('ignore_name_contains', None) - self.ignore_name = kwargs.get('ignore_name', None) or tuple() - self.include_apps = kwargs.get('include_apps', None) - self.exclude_apps = kwargs.get('exclude_apps', None) - self.database = kwargs.get('database', None) or 'default' - self.cache_path = kwargs.get('cache_path', None) or DEFAULT_CACHE_PATH - self.no_cache = kwargs.get('no_cache', None) or False + self.ignore_name_contains = kwargs.get("ignore_name_contains", None) + self.ignore_name = kwargs.get("ignore_name", None) or tuple() + self.include_apps = kwargs.get("include_apps", None) + self.exclude_apps = kwargs.get("exclude_apps", None) + self.database = kwargs.get("database", None) or "default" + self.cache_path = kwargs.get("cache_path", None) or DEFAULT_CACHE_PATH + self.no_cache = kwargs.get("no_cache", None) or False - self.python_exe = '{0}/bin/{1}'.format(sys.prefix, 'python') if \ - hasattr(sys, 'real_prefix') else 'python' + self.python_exe = ( + "{0}/bin/{1}".format(sys.prefix, "python") + if hasattr(sys, "real_prefix") + else "python" + ) # Initialise counters self.nb_valid = 0 @@ -68,65 +73,62 @@ class MigrationLinter(object): self.old_cache.load() def lint_migration(self, app_name, migration_name): - print('({0}, {1})... '.format(app_name, migration_name), end='') + print("({0}, {1})... ".format(app_name, migration_name), end="") self.nb_total += 1 md5hash = self.old_cache.md5(app_name, migration_name) if md5hash in self.old_cache: - if self.old_cache[md5hash]['result'] == 'IGNORE': - print('IGNORE (cached)') + if self.old_cache[md5hash]["result"] == "IGNORE": + print("IGNORE (cached)") self.nb_ignored += 1 - elif self.old_cache[md5hash]['result'] == 'OK': - print('OK (cached)') + elif self.old_cache[md5hash]["result"] == "OK": + print("OK (cached)") self.nb_valid += 1 else: - print('ERR (cached)') + print("ERR (cached)") self.nb_erroneous += 1 - if 'errors' in self.old_cache[md5hash]: - self.print_errors(self.old_cache[md5hash]['errors']) + if "errors" in self.old_cache[md5hash]: + self.print_errors(self.old_cache[md5hash]["errors"]) self.new_cache[md5hash] = self.old_cache[md5hash] return if self.should_ignore_migration(app_name, migration_name): - print('IGNORE') + print("IGNORE") self.nb_ignored += 1 return - sql_statements = self.get_sql( - app_name, migration_name) - analysis_result = analyse_sql_statements( - sql_statements) - errors = analysis_result['errors'] + sql_statements = self.get_sql(app_name, migration_name) + analysis_result = analyse_sql_statements(sql_statements) + errors = analysis_result["errors"] - if analysis_result['ignored']: - print('IGNORE') - self.new_cache[md5hash] = {'result': 'IGNORE'} + if analysis_result["ignored"]: + print("IGNORE") + self.new_cache[md5hash] = {"result": "IGNORE"} self.nb_ignored += 1 return if not errors: - print('OK') - self.new_cache[md5hash] = {'result': 'OK'} + print("OK") + self.new_cache[md5hash] = {"result": "OK"} self.nb_valid += 1 return - print('ERR') - self.new_cache[md5hash] = {'result': 'ERR', 'errors': errors} + print("ERR") + self.new_cache[md5hash] = {"result": "ERR", "errors": errors} self.nb_erroneous += 1 self.print_errors(errors) @staticmethod def print_errors(errors): for err in errors: - error_str = '\t{0}'.format(err['err_msg']) - if err['table']: - error_str += ' (table: {0}'.format(err['table']) - if err['column']: - error_str += ', column: {0}'.format( - err['column']) - error_str += ')' + error_str = "\t{0}".format(err["err_msg"]) + if err["table"]: + error_str += " (table: {0}".format(err["table"]) + if err["column"]: + error_str += ", column: {0}".format(err["column"]) + error_str += ")" print(error_str) def lint_all_migrations(self, git_commit_id=None): @@ -134,8 +136,10 @@ class MigrationLinter(object): if git_commit_id: if not utils.is_git_project(self.django_path): raise ValueError( - ('The given project {0} does not seem ' - 'to be versioned by git.').format(self.django_path)) + ( + "The given project {0} does not seem " "to be versioned by git." + ).format(self.django_path) + ) migrations = self._gather_migrations_git(git_commit_id) else: migrations = self._gather_all_migrations() @@ -148,14 +152,14 @@ class MigrationLinter(object): self.new_cache.save() def print_summary(self): - print('*** Summary:') - print(('Valid migrations: {1}/{0} - ' - 'erroneous migrations: {2}/{0} - ' - 'ignored migrations: {3}/{0}').format( - self.nb_total, - self.nb_valid, - self.nb_erroneous, - self.nb_ignored)) + print("*** Summary:") + print( + ( + "Valid migrations: {1}/{0} - " + "erroneous migrations: {2}/{0} - " + "ignored migrations: {3}/{0}" + ).format(self.nb_total, self.nb_valid, self.nb_erroneous, self.nb_ignored) + ) @property def has_errors(self): @@ -174,153 +178,161 @@ class MigrationLinter(object): it allows to seperate the instances correctly. """ sqlmigrate_command = ( - 'cd {0} && ' - '{1} manage.py sqlmigrate {2} {3} ' - '--database {4}').format( - self.django_path, - self.python_exe, app_name, migration_name, - self.database) - logger.info('Executing {0}'.format(sqlmigrate_command)) + "cd {0} && " "{1} manage.py sqlmigrate {2} {3} " "--database {4}" + ).format( + self.django_path, self.python_exe, app_name, migration_name, self.database + ) + logger.info("Executing {0}".format(sqlmigrate_command)) sqlmigrate_process = Popen( - sqlmigrate_command, shell=True, stdout=PIPE, stderr=PIPE) + sqlmigrate_command, shell=True, stdout=PIPE, stderr=PIPE + ) sql_statements = [] for line in map( - utils.clean_bytes_to_str, - sqlmigrate_process.stdout.readlines()): + utils.clean_bytes_to_str, sqlmigrate_process.stdout.readlines() + ): sql_statements.append(line) sqlmigrate_process.wait() if sqlmigrate_process.returncode != 0: _, err = sqlmigrate_process.communicate() - raise RuntimeError('sqlmigrate command failed {0}'.format( - err.decode('utf-8'))) - logger.info( - 'Found {0} sql migration lines'.format(len(sql_statements))) + raise RuntimeError( + "sqlmigrate command failed {0}".format(err.decode("utf-8")) + ) + logger.info("Found {0} sql migration lines".format(len(sql_statements))) return sql_statements def _gather_migrations_git(self, git_commit_id): migrations = [] # Get changes since specified commit git_diff_command = ( - 'cd {0} && ' - 'git diff --name-only --diff-filter=A {1}').format( - self.django_path, git_commit_id) - logger.info('Executing {0}'.format(git_diff_command)) - diff_process = Popen( - git_diff_command, shell=True, stdout=PIPE, stderr=PIPE) - for line in map( - utils.clean_bytes_to_str, diff_process.stdout.readlines()): + "cd {0} && " "git diff --name-only --diff-filter=A {1}" + ).format(self.django_path, git_commit_id) + logger.info("Executing {0}".format(git_diff_command)) + diff_process = Popen(git_diff_command, shell=True, stdout=PIPE, stderr=PIPE) + for line in map(utils.clean_bytes_to_str, diff_process.stdout.readlines()): # Only gather lines that include added migrations - if re.search( - r'\/{0}\/.*\.py'.format(MIGRATION_FOLDER_NAME), - line) and \ - '__init__' not in line: + if ( + re.search(r"\/{0}\/.*\.py".format(MIGRATION_FOLDER_NAME), line) + and "__init__" not in line + ): app_name, migration_name = split_migration_path(line) migrations.append((app_name, migration_name)) diff_process.wait() if diff_process.returncode != 0: output = [] - for line in map( - utils.clean_bytes_to_str, diff_process.stderr.readlines()): + for line in map(utils.clean_bytes_to_str, diff_process.stderr.readlines()): output.append(line) - logger.info("Error while git diff command:\n{}".format( - "".join(output))) - raise Exception('Error while executing git diff command') + logger.info("Error while git diff command:\n{}".format("".join(output))) + raise Exception("Error while executing git diff command") return migrations def _gather_all_migrations(self): migrations = [] for root, dirs, files in os.walk(self.django_path): for file_name in sorted(files): - if os.path.basename(root) == MIGRATION_FOLDER_NAME and \ - file_name.endswith('.py') and \ - file_name != '__init__.py': + if ( + os.path.basename(root) == MIGRATION_FOLDER_NAME + and file_name.endswith(".py") + and file_name != "__init__.py" + ): full_migration_path = os.path.join(root, file_name) - app_name, migration_name = split_migration_path( - full_migration_path) + app_name, migration_name = split_migration_path(full_migration_path) migrations.append((app_name, migration_name)) return migrations def should_ignore_migration(self, app_name, migration_name): - return (self.include_apps and - app_name not in self.include_apps)\ - or (self.exclude_apps and - app_name in self.exclude_apps)\ - or (self.ignore_name_contains and - self.ignore_name_contains in migration_name)\ + return ( + (self.include_apps and app_name not in self.include_apps) + or (self.exclude_apps and app_name in self.exclude_apps) + or ( + self.ignore_name_contains + and self.ignore_name_contains in migration_name + ) or (migration_name in self.ignore_name) + ) def _main(): import argparse + parser = argparse.ArgumentParser( - description='Detect backward incompatible django migrations.') + description="Detect backward incompatible django migrations." + ) parser.add_argument( - 'django_folder', - metavar='DJANGO_FOLDER', + "django_folder", + metavar="DJANGO_FOLDER", type=str, nargs=1, - help='the path to the django project') + help="the path to the django project", + ) parser.add_argument( - 'commit_id', - metavar='GIT_COMMIT_ID', - type=str, nargs='?', - help=('if specified, only migrations since this commit ' - 'will be taken into account. If not specified, ' - 'the initial repo commit will be used')) - parser.add_argument( - '--ignore-name-contains', + "commit_id", + metavar="GIT_COMMIT_ID", type=str, - nargs='?', - help='ignore migrations containing this name') + nargs="?", + help=( + "if specified, only migrations since this commit " + "will be taken into account. If not specified, " + "the initial repo commit will be used" + ), + ) parser.add_argument( - '--ignore-name', + "--ignore-name-contains", type=str, - nargs='*', - help='ignore migrations with exactly one of these names') + nargs="?", + help="ignore migrations containing this name", + ) parser.add_argument( - '--verbose', - '-v', - action='store_true', - help='print more information during execution') - parser.add_argument( - '--database', + "--ignore-name", type=str, - nargs='?', - help=('specify the database for which to generate the SQL. ' - 'Defaults to default')) + nargs="*", + help="ignore migrations with exactly one of these names", + ) + parser.add_argument( + "--verbose", + "-v", + action="store_true", + help="print more information during execution", + ) + parser.add_argument( + "--database", + type=str, + nargs="?", + help=( + "specify the database for which to generate the SQL. " "Defaults to default" + ), + ) - cache_group = parser.add_mutually_exclusive_group( - required=False) + cache_group = parser.add_mutually_exclusive_group(required=False) cache_group.add_argument( - '--cache-path', + "--cache-path", type=str, - help='specify a directory that should be used to' - 'store cache-files in.') + help="specify a directory that should be used to" "store cache-files in.", + ) cache_group.add_argument( - '--no-cache', - action='store_true', - help='don\'t use a cache') + "--no-cache", action="store_true", help="don't use a cache" + ) - incl_excl_group = parser.add_mutually_exclusive_group( - required=False) + incl_excl_group = parser.add_mutually_exclusive_group(required=False) incl_excl_group.add_argument( - '--include-apps', + "--include-apps", type=str, - nargs='*', - help='check only migrations that are in the specified django apps') + nargs="*", + help="check only migrations that are in the specified django apps", + ) incl_excl_group.add_argument( - '--exclude-apps', + "--exclude-apps", type=str, - nargs='*', - help='ignore migrations that are in the specified django apps') + nargs="*", + help="ignore migrations that are in the specified django apps", + ) args = parser.parse_args() if args.verbose: - logging.basicConfig(format='%(message)s', level=logging.DEBUG) + logging.basicConfig(format="%(message)s", level=logging.DEBUG) else: - logging.basicConfig(format='%(message)s') + logging.basicConfig(format="%(message)s") folder_name = args.django_folder[0] # Create and use linter @@ -332,14 +344,13 @@ def _main(): exclude_apps=args.exclude_apps, database=args.database, cache_path=args.cache_path, - no_cache=args.no_cache + no_cache=args.no_cache, ) - linter.lint_all_migrations( - git_commit_id=args.commit_id) + linter.lint_all_migrations(git_commit_id=args.commit_id) linter.print_summary() if linter.has_errors: sys.exit(1) -if __name__ == '__main__': +if __name__ == "__main__": _main() diff --git a/django_migration_linter/operations.py b/django_migration_linter/operations.py index cf98135..cea05c0 100644 --- a/django_migration_linter/operations.py +++ b/django_migration_linter/operations.py @@ -26,12 +26,10 @@ class IgnoreMigration(Operation): def state_forwards(self, app_label, state): pass - def database_forwards( - self, app_label, schema_editor, from_state, to_state): + def database_forwards(self, app_label, schema_editor, from_state, to_state): schema_editor.execute(IGNORE_MIGRATION_SQL) - def database_backwards( - self, app_label, schema_editor, from_state, to_state): + def database_backwards(self, app_label, schema_editor, from_state, to_state): schema_editor.execute(IGNORE_MIGRATION_SQL + "(reversed)") def describe(self): diff --git a/django_migration_linter/sql_analyser.py b/django_migration_linter/sql_analyser.py index fa0a0d9..4e13244 100644 --- a/django_migration_linter/sql_analyser.py +++ b/django_migration_linter/sql_analyser.py @@ -23,55 +23,58 @@ logger = logging.getLogger(__name__) def has_default(sql, **kwargs): - if re.search('SET DEFAULT', sql) and kwargs['errors']: - err = next( - (err for err in kwargs['errors'] if err['code'] == 'NOT_NULL'), - None, - ) + if re.search("SET DEFAULT", sql) and kwargs["errors"]: + err = next((err for err in kwargs["errors"] if err["code"] == "NOT_NULL"), None) if err: logger.info( - ('Found a NOT_NULL error in migration, ' - 'but it has a default value added: {}').format(err) + ( + "Found a NOT_NULL error in migration, " + "but it has a default value added: {}" + ).format(err) ) - kwargs['errors'].remove(err) + kwargs["errors"].remove(err) return False # Never fails migration_tests = ( { - 'code': 'NOT_NULL', - 'fn': lambda sql, **kw: re.search('NOT NULL', sql) and - not re.search('CREATE TABLE', sql), - 'err_msg': 'NOT NULL constraint on columns' - }, { - 'code': 'DROP_COLUMN', - 'fn': lambda sql, **kw: re.search('DROP COLUMN', sql), - 'err_msg': 'DROPPING columns' - }, { - 'code': 'RENAME_COLUMN', - 'fn': lambda sql, **kw: re.search('ALTER TABLE .* CHANGE', sql) or - re.search('ALTER TABLE .* RENAME COLUMN', sql), - 'err_msg': 'RENAMING columns' - }, { - 'code': 'RENAME_TABLE', - 'fn': lambda sql, **kw: re.search('RENAME TABLE', sql) or - re.search('ALTER TABLE .* RENAME TO', sql), - 'err_msg': 'RENAMING tables' - }, { - 'code': 'ALTER_COLUMN', - 'fn': lambda sql, **kw: re.search('ALTER TABLE .* MODIFY', sql), - 'err_msg': ('ALTERING columns (Could be backward compatible. ' - 'You may ignore this migration.)') - }, { - 'code': '', - 'fn': has_default, - 'err_msg': '' - }, { - 'code': IGNORED_MIGRATION, - 'fn': lambda sql, **kw: re.search(IGNORE_MIGRATION_SQL, sql), - 'err_msg': '', - } + "code": "NOT_NULL", + "fn": lambda sql, **kw: re.search("NOT NULL", sql) + and not re.search("CREATE TABLE", sql), + "err_msg": "NOT NULL constraint on columns", + }, + { + "code": "DROP_COLUMN", + "fn": lambda sql, **kw: re.search("DROP COLUMN", sql), + "err_msg": "DROPPING columns", + }, + { + "code": "RENAME_COLUMN", + "fn": lambda sql, **kw: re.search("ALTER TABLE .* CHANGE", sql) + or re.search("ALTER TABLE .* RENAME COLUMN", sql), + "err_msg": "RENAMING columns", + }, + { + "code": "RENAME_TABLE", + "fn": lambda sql, **kw: re.search("RENAME TABLE", sql) + or re.search("ALTER TABLE .* RENAME TO", sql), + "err_msg": "RENAMING tables", + }, + { + "code": "ALTER_COLUMN", + "fn": lambda sql, **kw: re.search("ALTER TABLE .* MODIFY", sql), + "err_msg": ( + "ALTERING columns (Could be backward compatible. " + "You may ignore this migration.)" + ), + }, + {"code": "", "fn": has_default, "err_msg": ""}, + { + "code": IGNORED_MIGRATION, + "fn": lambda sql, **kw: re.search(IGNORE_MIGRATION_SQL, sql), + "err_msg": "", + }, ) @@ -80,32 +83,23 @@ def analyse_sql_statements(sql_statements): ignored = False for statement in sql_statements: for test in migration_tests: - if test['fn'](statement, errors=errors): - if test['code'] == IGNORED_MIGRATION: - logger.info( - 'Testing {0} -- IGNORING MIGRATION'.format(statement) - ) + if test["fn"](statement, errors=errors): + if test["code"] == IGNORED_MIGRATION: + logger.info("Testing {0} -- IGNORING MIGRATION".format(statement)) ignored = True else: - logger.info('Testing {0} -- ERROR'.format(statement)) + logger.info("Testing {0} -- ERROR".format(statement)) table_search = re.search( - 'TABLE `([^`]*)`', statement, re.IGNORECASE) - col_search = re.search( - 'COLUMN `([^`]*)`', statement, re.IGNORECASE) + "TABLE `([^`]*)`", statement, re.IGNORECASE + ) + col_search = re.search("COLUMN `([^`]*)`", statement, re.IGNORECASE) err = { - 'err_msg': - test['err_msg'], - 'code': - test['code'], - 'table': - table_search.group(1) if table_search else None, - 'column': - col_search.group(1) if col_search else None + "err_msg": test["err_msg"], + "code": test["code"], + "table": table_search.group(1) if table_search else None, + "column": col_search.group(1) if col_search else None, } errors.append(err) else: - logger.info('Testing {0} -- PASSED'.format(statement)) - return { - 'errors': errors, - 'ignored': ignored - } + logger.info("Testing {0} -- PASSED".format(statement)) + return {"errors": errors, "ignored": ignored} diff --git a/django_migration_linter/utils.py b/django_migration_linter/utils.py index 355df5f..d3a88d6 100644 --- a/django_migration_linter/utils.py +++ b/django_migration_linter/utils.py @@ -21,12 +21,12 @@ from django_migration_linter.constants import MIGRATION_FOLDER_NAME def is_django_project(path): - django_manage_file = os.path.join(path, 'manage.py') + django_manage_file = os.path.join(path, "manage.py") return os.path.isfile(django_manage_file) def is_git_project(path): - git_directory = os.path.join(path, '.git') + git_directory = os.path.join(path, ".git") return os.path.isdir(git_directory) @@ -44,33 +44,31 @@ def find_project_settings_module(path): """ for root, dirs, files in os.walk(path): for file_name in files: - if file_name == 'settings.py': - return os.path.join( - root.replace(path, ''), - file_name).replace('/', '.').rstrip('.py') + if file_name == "settings.py": + return ( + os.path.join(root.replace(path, ""), file_name) + .replace("/", ".") + .rstrip(".py") + ) def split_path(path): a, b = os.path.split(path) - return (split_path(a) if (len(a) > 0 and a != '/') else []) + [b] + return (split_path(a) if (len(a) > 0 and a != "/") else []) + [b] def split_migration_path(migration_path): decomposed_path = split_path(migration_path) for i, p in enumerate(decomposed_path): if p == MIGRATION_FOLDER_NAME: - return (decomposed_path[i-1], - os.path.splitext(decomposed_path[i+1])[0]) + return (decomposed_path[i - 1], os.path.splitext(decomposed_path[i + 1])[0]) def compose_migration_path(django_folder, app_name, migration): return os.path.join( - django_folder, - app_name, - MIGRATION_FOLDER_NAME, - '{0}.py'.format(migration) + django_folder, app_name, MIGRATION_FOLDER_NAME, "{0}.py".format(migration) ) def clean_bytes_to_str(byte_input): - return byte_input.decode('utf-8').strip() + return byte_input.decode("utf-8").strip()