public/django-migration-linter
15
0 Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Linter Core: change lots of behaviour

Browse Source 
* Use the Django migration classes to manipulate these objects
this includes the discovery of migrations through the MigrationLoader
(instead of manually walking through the Django project and find the migrations through a regex)
and representing a Migration through the Django class.
* Get the SQL of a migration through a 'call_command' call of 'sqlmigrate'
That speeds up greatly the linter since we don't have to setup Django for each migration
* Add the database to the cache path
This commit is contained in:
David Wobrock committed 2019-05-12 15:32:57 +02:00
1 parent 0eb584889a
commit 484a775fda
4 files changed
+122 -272
Download Patch File Download Diff File Expand all files Collapse all files

No files matched your search

django_migration_linter/cache.py
+5 -6
View File
@@ -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))
django_migration_linter/migration.py
-21
View File
@@ -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()
django_migration_linter/migration_linter.py
+100 -208
View File
@@ -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()
django_migration_linter/utils.py
+17 -37
View File
@@ -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