class Cache(dict):
    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))

        super(Cache, self).__init__()

    def load(self):
        try:
            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:
            pickle.dump(self, f, protocol=2) DEFAULT_CACHE_PATH = user_cache_dir("django-migration-linter", version=__version__) DJANGO_APPS_WITH_MIGRATIONS = ("admin", "auth", "contenttypes", "sessions") EXPECTED_DATA_MIGRATION_ARGS = ("apps", "schema_editor") class IgnoreMigration(Operation):
    """
    No-op migration operation that will enable the Django Migration Linter
    to detect if the entire migration should be ignored (through code).
    """

    reversible = True
    reduces_to_sql = False

    def state_forwards(self, app_label, state):
        pass

    def database_forwards(self, app_label, schema_editor, from_state, to_state):
        pass

    def database_backwards(self, app_label, schema_editor, from_state, to_state):
        pass

    def describe(self):
        return "The Django migration linter will ignore this migration" def analyse_sql_statements(
    sql_analyser_class, sql_statements, exclude_migration_tests=None
):
    sql_analyser = sql_analyser_class(exclude_migration_tests)
    sql_analyser.analyse(sql_statements)
    return sql_analyser.errors, sql_analyser.ignored, sql_analyser.warnings def get_sql_analyser_class(database_vendor, analyser_string=None):
    if analyser_string:
        return get_sql_analyser_from_string(analyser_string)
    return get_sql_analyser_class_from_db_vendor(database_vendor) 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 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 == MIGRATIONS_MODULE_NAME:
            return decomposed_path[i - 1], os.path.splitext(decomposed_path[i + 1])[0]
import hashlib
import inspect
import logging
import os
import re
from enum import Enum, unique
from subprocess import PIPE, Popen
from django.conf import settings
from django.core.management import call_command
from django.db import DEFAULT_DB_ALIAS, ProgrammingError, connections
from django.db.migrations import RunPython, RunSQL
from .cache import Cache
from .constants import (
    DEFAULT_CACHE_PATH,
    DJANGO_APPS_WITH_MIGRATIONS,
    EXPECTED_DATA_MIGRATION_ARGS,
)
from .operations import IgnoreMigration
from .sql_analyser import analyse_sql_statements, get_sql_analyser_class
from .utils import clean_bytes_to_str, get_migration_abspath, split_migration_path
        from django.db.migrations.loader import MigrationLoader
        from django.db.migrations.loader import MIGRATIONS_MODULE_NAME

    def lint_cached_migration(self, app_label, migration_name, md5hash):
        cached_value = self.old_cache[md5hash]
        if cached_value["result"] == "IGNORE":
            self.print_linting_msg(
                app_label, migration_name, "IGNORE (cached)", MessageType.IGNORE
            )
            self.nb_ignored += 1
        elif cached_value["result"] == "OK":
            self.print_linting_msg(
                app_label, migration_name, "OK (cached)", MessageType.OK
            )
            self.nb_valid += 1
        elif cached_value["result"] == "WARNING":
            self.print_linting_msg(
                app_label, migration_name, "WARNING (cached)", MessageType.WARNING
            )
            self.nb_warnings += 1
            self.print_warnings(cached_value["warnings"])
        else:
            self.print_linting_msg(
                app_label, migration_name, "ERR (cached)", MessageType.ERROR
            )
            self.nb_erroneous += 1
            if "errors" in cached_value:
                self.print_errors(cached_value["errors"])
            if "warnings" in cached_value and cached_value["warnings"]:
                self.print_warnings(cached_value["warnings"])

        self.new_cache[md5hash] = cached_value

    def print_linting_msg(self, app_label, migration_name, msg, lint_result):
        if lint_result.value in self.quiet:
            return
        if not self.no_output:
            print("({0}, {1})... {2}".format(app_label, migration_name, msg))

    def print_errors(self, errors):
        if MessageType.ERROR.value in self.quiet:
            return
        for err in errors:
            error_str = "\t{0}".format(err["msg"])
            if err.get("table"):
                error_str += " (table: {0}".format(err["table"])
                if err.get("column"):
                    error_str += ", column: {0}".format(err["column"])
                error_str += ")"
            if not self.no_output:
                print(error_str)

    def print_warnings(self, warnings):
        if MessageType.WARNING.value in self.quiet:
            return

        for warning_details in warnings:
            warn_str = "\t{}".format(warning_details["msg"])
            if not self.no_output:
                print(warn_str)

    def print_summary(self):
        if self.no_output:
            return
        print("*** Summary ***")
        print("Valid migrations: {}/{}".format(self.nb_valid, self.nb_total))
        print("Erroneous migrations: {}/{}".format(self.nb_erroneous, self.nb_total))
        print("Migrations with warnings: {}/{}".format(self.nb_warnings, self.nb_total))
        print("Ignored migrations: {}/{}".format(self.nb_ignored, self.nb_total))

    @property
    def has_errors(self):
        return self.nb_erroneous > 0

    def get_sql(self, app_label, migration_name):
        logger.info(
            "Calling sqlmigrate command {} {}".format(app_label, migration_name)
        )
        dev_null = open(os.devnull, "w")
        try:
            sql_statement = call_command(
                "sqlmigrate",
                app_label,
                migration_name,
                database=self.database,
                stdout=dev_null,
            )
        except (ValueError, ProgrammingError):
            logger.warning(
                (
                    "Error while executing sqlmigrate on (%s, %s). "
                    "Continuing execution with empty SQL."
                ),
                app_label,
                migration_name,
            )
            sql_statement = ""
        return sql_statement.splitlines()

    @staticmethod
    def is_migration_file(filename):

        return (
            re.search(r"/{0}/.*\.py".format(MIGRATIONS_MODULE_NAME), filename)
            and "__init__" not in filename
        )

    @classmethod
    def read_migrations_list(cls, migrations_file_path):
        """
        Returning an empty list is different from returning None here.
        None: no file was specified and we should consider all migrations
        Empty list: no migration found in the file and we should consider no migration
        """
        if not migrations_file_path:
            return None

        migrations = []
        try:
            with open(migrations_file_path, "r") as file:
                for line in file:
                    if cls.is_migration_file(line):

                        app_label, name = split_migration_path(line)