diff --git a/django_migration_linter/cache.py b/django_migration_linter/cache.py index d64ef3e..e85b072 100644 --- a/django_migration_linter/cache.py +++ b/django_migration_linter/cache.py @@ -15,14 +15,13 @@ import os import pickle -from .utils import split_path - class Cache(dict): - def __init__(self, django_folder, cache_path): - self.django_folder = django_folder - project_name = split_path(django_folder)[-1] - self.filename = os.path.join(cache_path, "{0}.pickle".format(project_name)) + def __init__(self, django_folder, database, cache_path): + self.filename = os.path.join( + cache_path, + "{0}_{1}.pickle".format(django_folder.replace(os.sep, "_"), database), + ) if not os.path.exists(os.path.dirname(self.filename)): os.makedirs(os.path.dirname(self.filename)) diff --git a/django_migration_linter/migration.py b/django_migration_linter/migration.py deleted file mode 100644 index c24a2ae..0000000 --- a/django_migration_linter/migration.py +++ /dev/null @@ -1,21 +0,0 @@ -import hashlib - -from django_migration_linter.utils import split_migration_path - - -class Migration(object): - def __init__(self, abs_path): - self.abs_path = abs_path - self.app_name, self.name = split_migration_path(self.abs_path) - - def __lt__(self, other): - if not isinstance(other, Migration): - return NotImplemented - return self.abs_path < other.abs_path - - def get_md5hash(self): - hash_md5 = hashlib.md5() - with open(self.abs_path, "rb") as f: - for chunk in iter(lambda: f.read(4096), b""): - hash_md5.update(chunk) - return hash_md5.hexdigest() diff --git a/django_migration_linter/migration_linter.py b/django_migration_linter/migration_linter.py index 3961f23..efb900f 100644 --- a/django_migration_linter/migration_linter.py +++ b/django_migration_linter/migration_linter.py @@ -13,46 +13,48 @@ # limitations under the License. from __future__ import print_function + +import hashlib import logging import os import re from subprocess import Popen, PIPE -import sys + +from django.core.management import call_command +from django.db import DEFAULT_DB_ALIAS +from django.db.migrations import Migration from .cache import Cache -from .constants import DEFAULT_CACHE_PATH, MIGRATION_FOLDER_NAME, __version__ -from .migration import Migration -from .utils import is_directory, is_django_project, clean_bytes_to_str +from .constants import DEFAULT_CACHE_PATH +from .utils import clean_bytes_to_str, get_migration_abspath, split_migration_path from .sql_analyser import analyse_sql_statements logger = logging.getLogger(__name__) +DJANGO_APPS_WITH_MIGRATIONS = ("admin", "auth", "contenttypes", "sessions") + class MigrationLinter(object): - def __init__(self, project_path, **kwargs): - # Verify correctness - if not is_directory(project_path): - raise ValueError( - "The given path {0} does not seem to be a directory.".format( - project_path - ) - ) - if not is_django_project(project_path): - raise ValueError( - "The given path {0} does not seem to be a django project.".format( - project_path - ) - ) - + def __init__( + self, + path=None, + ignore_name_contains=None, + ignore_name=None, + include_apps=None, + exclude_apps=None, + database=DEFAULT_DB_ALIAS, + cache_path=DEFAULT_CACHE_PATH, + no_cache=False, + ): # Store parameters and options - self.django_path = os.path.abspath(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.django_path = path + self.ignore_name_contains = ignore_name_contains + self.ignore_name = ignore_name or tuple() + self.include_apps = include_apps + self.exclude_apps = exclude_apps + self.database = database or DEFAULT_DB_ALIAS + self.cache_path = cache_path or DEFAULT_CACHE_PATH + self.no_cache = no_cache # Initialise counters self.nb_valid = 0 @@ -61,51 +63,80 @@ class MigrationLinter(object): self.nb_total = 0 # Initialise cache. Read from old, write to new to prune old entries. - if not self.no_cache: - self.old_cache = Cache(self.django_path, self.cache_path) - self.new_cache = Cache(self.django_path, self.cache_path) + if self.should_use_cache(): + self.old_cache = Cache(self.django_path, self.database, self.cache_path) + self.new_cache = Cache(self.django_path, self.database, self.cache_path) self.old_cache.load() + def should_use_cache(self): + return self.django_path and not self.no_cache + + def lint_all_migrations(self, git_commit_id=None): + # Collect migrations + if git_commit_id: + migrations = self._gather_migrations_git(git_commit_id) + else: + migrations = self._gather_all_migrations() + + # Lint those migrations + sorted_migrations = sorted( + migrations, key=lambda migration: (migration.app_label, migration.name) + ) + for m in sorted_migrations: + self.lint_migration(m) + + if self.should_use_cache(): + self.new_cache.save() + def lint_migration(self, migration): - app_name = migration.app_name + app_label = migration.app_label migration_name = migration.name - print("({0}, {1})... ".format(app_name, migration_name), end="") + print("({0}, {1})... ".format(app_label, migration_name), end="") self.nb_total += 1 - md5hash = migration.get_md5hash() - if not self.no_cache and md5hash in self.old_cache: + md5hash = self.get_migration_hash(app_label, migration_name) + + if self.should_use_cache() and md5hash in self.old_cache: self.lint_cached_migration(md5hash) return - if self.should_ignore_migration(app_name, migration_name): + if self.should_ignore_migration(app_label, migration_name): print("IGNORE") self.nb_ignored += 1 return - sql_statements = self.get_sql(app_name, migration_name) + sql_statements = self.get_sql(app_label, migration_name) analysis_result = analyse_sql_statements(sql_statements) errors = analysis_result["errors"] if analysis_result["ignored"]: print("IGNORE") self.nb_ignored += 1 - if not self.no_cache: + if self.should_use_cache(): self.new_cache[md5hash] = {"result": "IGNORE"} return if not errors: print("OK") self.nb_valid += 1 - if not self.no_cache: + if self.should_use_cache(): self.new_cache[md5hash] = {"result": "OK"} return print("ERR") self.nb_erroneous += 1 self.print_errors(errors) - if not self.no_cache: + if self.should_use_cache(): self.new_cache[md5hash] = {"result": "ERR", "errors": errors} + @staticmethod + def get_migration_hash(app_label, migration_name): + hash_md5 = hashlib.md5() + with open(get_migration_abspath(app_label, migration_name), "rb") as f: + for chunk in iter(lambda: f.read(4096), b""): + hash_md5.update(chunk) + return hash_md5.hexdigest() + def lint_cached_migration(self, md5hash): cached_value = self.old_cache[md5hash] if cached_value["result"] == "IGNORE": @@ -133,20 +164,6 @@ class MigrationLinter(object): error_str += ")" print(error_str) - def lint_all_migrations(self, git_commit_id=None): - # Collect migrations - if git_commit_id: - migrations = self._gather_migrations_git(git_commit_id) - else: - migrations = self._gather_all_migrations() - - # Lint those migrations - for m in migrations: - self.lint_migration(m) - - if not self.no_cache: - self.new_cache.save() - def print_summary(self): print("*** Summary:") print( @@ -161,55 +178,38 @@ class MigrationLinter(object): def has_errors(self): return self.nb_erroneous > 0 - def get_sql(self, app_name, migration_name): - """ It would be much faster to call the - command directly from the code using - call_command(), but requires the code - to setup django (by calling django.setup()) - and set the DJANGO_SETTINGS_MODULE. - However, django is global and doesn't allow - multiple linter instances to exist at the same time. - (because they would all just lint the same django project) - Even if calling a shell is slow and ugly, for now, - it allows to seperate the instances correctly. - """ - sqlmigrate_command = ( - "cd {0} && {1} manage.py sqlmigrate {2} '{3}' --database {4}" - ).format( - self.django_path, sys.executable, app_name, migration_name, self.database + def get_sql(self, app_label, migration_name): + logger.info( + "Calling sqlmigrate command {} {}".format(app_label, migration_name) ) - logger.info("Executing {0}".format(sqlmigrate_command)) - sqlmigrate_process = Popen( - sqlmigrate_command, shell=True, stdout=PIPE, stderr=PIPE + dev_null = open(os.devnull, "w") + sql_statement = call_command( + "sqlmigrate", + app_label, + migration_name, + database=self.database, + stdout=dev_null, ) - - sql_statements = [] - for line in map(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))) - return sql_statements + return sql_statement.splitlines() def _gather_migrations_git(self, git_commit_id): + from django.db.migrations.loader import MIGRATIONS_MODULE_NAME + migrations = [] # Get changes since specified commit git_diff_command = ( - "cd {0} && git diff --relative --name-only --diff-filter=A {1}" + "cd {0} && git diff --relative --name-only --diff-filter=AR {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(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) + re.search(r"/{0}/.*\.py".format(MIGRATIONS_MODULE_NAME), line) and "__init__" not in line ): - migrations.append(Migration(os.path.join(self.django_path, line))) + app_label, name = split_migration_path(line) + migrations.append(Migration(name, app_label)) diff_process.wait() if diff_process.returncode != 0: @@ -220,132 +220,24 @@ class MigrationLinter(object): 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" - ): - full_migration_path = os.path.join(root, file_name) - migrations.append(Migration(full_migration_path)) - return migrations + @staticmethod + def _gather_all_migrations(): + from django.db.migrations.loader import MigrationLoader - def should_ignore_migration(self, app_name, migration_name): + migration_loader = MigrationLoader(connection=None, load=False) + migration_loader.load_disk() + # Prune Django apps + for (app_label, _), migration in migration_loader.disk_migrations.items(): + if app_label not in DJANGO_APPS_WITH_MIGRATIONS: + yield migration + + def should_ignore_migration(self, app_label, 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) + (self.include_apps and app_label not in self.include_apps) + or (self.exclude_apps and app_label 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." - ) - parser.add_argument( - "django_folder", - metavar="DJANGO_FOLDER", - type=str, - nargs=1, - 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", - type=str, - nargs="?", - help="ignore migrations containing this name", - ) - parser.add_argument( - "--ignore-name", - type=str, - 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( - "--version", "-V", action="version", version="%(prog)s {}".format(__version__) - ) - 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.add_argument( - "--cache-path", - type=str, - 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" - ) - - incl_excl_group = parser.add_mutually_exclusive_group(required=False) - incl_excl_group.add_argument( - "--include-apps", - type=str, - nargs="*", - help="check only migrations that are in the specified django apps", - ) - incl_excl_group.add_argument( - "--exclude-apps", - type=str, - 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) - else: - logging.basicConfig(format="%(message)s") - - folder_name = args.django_folder[0] - # Create and use linter - linter = MigrationLinter( - folder_name, - ignore_name_contains=args.ignore_name_contains, - ignore_name=args.ignore_name, - include_apps=args.include_apps, - exclude_apps=args.exclude_apps, - database=args.database, - cache_path=args.cache_path, - no_cache=args.no_cache, - ) - linter.lint_all_migrations(git_commit_id=args.commit_id) - linter.print_summary() - if linter.has_errors: - sys.exit(1) - - -if __name__ == "__main__": - _main() diff --git a/django_migration_linter/utils.py b/django_migration_linter/utils.py index b398f76..34a6a17 100644 --- a/django_migration_linter/utils.py +++ b/django_migration_linter/utils.py @@ -15,36 +15,7 @@ from __future__ import print_function import os -import sys - -from .constants import MIGRATION_FOLDER_NAME - - -def is_django_project(path): - django_manage_file = os.path.join(path, "manage.py") - return os.path.isfile(django_manage_file) - - -def is_directory(path): - return os.path.isdir(path) - - -def print_stderr(*args, **kwargs): - print(*args, file=sys.stderr, **kwargs) - - -def find_project_settings_module(path): - """Explore path recursively to the first settings.py file - and translate the name to package notation 'mysite.settings' - """ - 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(os.sep, ".") - .rstrip(".py") - ) +from importlib import import_module def split_path(path): @@ -67,17 +38,26 @@ def split_path(path): def split_migration_path(migration_path): + from django.db.migrations.loader import MIGRATIONS_MODULE_NAME + decomposed_path = split_path(migration_path) for i, p in enumerate(decomposed_path): - if p == MIGRATION_FOLDER_NAME: + if p == MIGRATIONS_MODULE_NAME: 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) - ) - - def clean_bytes_to_str(byte_input): return byte_input.decode("utf-8").strip() + + +def get_migration_abspath(app_label, migration_name): + from django.db.migrations.loader import MigrationLoader + + module_name, _ = MigrationLoader.migrations_module(app_label) + migration_path = "{}.{}".format(module_name, migration_name) + migration_module = import_module(migration_path) + + migration_file = migration_module.__file__ + if migration_file.endswith(".pyc"): + migration_file = migration_file[:-1] + return migration_file