Black all files

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()