Refactor sql_analyser into multiple classes for simpler extension in the future

django_migration_linter/sql_analyser.py

@@ -1,196 +0,0 @@
-# Copyright 2019 3YOURMIND GmbH
-
-# Licensed under the Apache License, Version 2.0 (the "License");
-# you may not use this file except in compliance with the License.
-# You may obtain a copy of the License at
-
-# http://www.apache.org/licenses/LICENSE-2.0
-
-# Unless required by applicable law or agreed to in writing, software
-# distributed under the License is distributed on an "AS IS" BASIS,
-# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
-# See the License for the specific language governing permissions and
-# limitations under the License.
-
-import re
-import logging
-from copy import deepcopy
-
-logger = logging.getLogger(__name__)
-
-
-def find_error_dict_with_code(tests, code):
-    return next((test_dict for test_dict in tests if test_dict["code"] == code), None)
-
-
-def has_not_null_column(sql_statements, **kwargs):
-    # TODO: improve to detect that the same column is concerned
-    return any(
-        re.search("(?<!DROP )NOT NULL", sql) and not sql.startswith("CREATE TABLE")
-        for sql in sql_statements
-    ) and not any("SET DEFAULT" in sql for sql in sql_statements)
-
-
-base_migration_tests = [
-    {
-        "code": "RENAME_TABLE",
-        "fn": lambda sql, **kw: re.search("RENAME TABLE", sql)
-        or re.search("ALTER TABLE .* RENAME TO", sql),
-        "err_msg": "RENAMING tables",
-        "mode": "one_liner",
-    },
-    {
-        "code": "NOT_NULL",
-        "fn": has_not_null_column,
-        "err_msg": "NOT NULL constraint on columns",
-        "mode": "transaction",
-    },
-    {
-        "code": "DROP_COLUMN",
-        "fn": lambda sql, **kw: re.search("DROP COLUMN", sql),
-        "err_msg": "DROPPING columns",
-        "mode": "one_liner",
-    },
-    {
-        "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",
-        "mode": "one_liner",
-    },
-    {
-        "code": "ALTER_COLUMN",
-        "fn": lambda sql, **kw: re.search("ALTER TABLE .* ALTER COLUMN .* TYPE", sql),
-        "err_msg": (
-            "ALTERING columns (Could be backward compatible. "
-            "You may ignore this migration.)"
-        ),
-        "mode": "one_liner",
-    },
-]
-
-mysql_migration_tests = [
-    {
-        "code": "ALTER_COLUMN",
-        "fn": lambda sql, **kw: re.search("ALTER TABLE .* MODIFY .* (?!NULL);?$", sql),
-        "mode": "one_liner",
-    }
-]
-
-postgresql_migration_tests = []
-
-sqlite_migration_tests = [
-    {
-        "code": "RENAME_TABLE",
-        "fn": lambda sql, **kw: re.search("ALTER TABLE .* RENAME TO", sql)
-        and "__old" not in sql
-        and "new__" not in sql,
-    },
-    {
-        "code": "NOT_NULL",
-        "fn": lambda sql_statements, **kw: any(
-            re.search("NOT NULL(?! PRIMARY)(?! DEFAULT)", sql) for sql in sql_statements
-        )
-        and any(
-            re.search("ALTER TABLE .* RENAME TO", sql)
-            and ("__old" in sql or "new__" in sql)
-            for sql in sql_statements
-        ),
-        "mode": "transaction",
-    },
-]
-
-
-def update_migration_tests(base_tests, specific_tests):
-    for override_test in specific_tests:
-        migration_test_dict = find_error_dict_with_code(
-            base_tests, override_test["code"]
-        )
-
-        if migration_test_dict is None or not override_test["code"]:
-            migration_test_dict = {}
-            base_tests.append(migration_test_dict)
-
-        for key in override_test.keys():
-            migration_test_dict[key] = override_test[key]
-    return base_tests
-
-
-def get_migration_tests(database_vendor):
-    migration_tests = deepcopy(base_migration_tests)
-
-    if "mysql" in database_vendor:
-        migration_tests = update_migration_tests(migration_tests, mysql_migration_tests)
-    elif "postgre" in database_vendor:
-        migration_tests = update_migration_tests(
-            migration_tests, postgresql_migration_tests
-        )
-    elif "sqlite" in database_vendor:
-        migration_tests = update_migration_tests(
-            migration_tests, sqlite_migration_tests
-        )
-
-    return migration_tests
-
-
-def get_single_line_migration_tests(database_vendor):
-    return [
-        test
-        for test in get_migration_tests(database_vendor)
-        if test["mode"] == "one_liner"
-    ]
-
-
-def get_transaction_migration_tests(database_vendor):
-    # TODO: improve transaction to detect BEGIN & COMMIT
-    return [
-        test
-        for test in get_migration_tests(database_vendor)
-        if test["mode"] == "transaction"
-    ]
-
-
-def build_error_dict(migration_test, sql_statement=""):
-    table_search = re.search("TABLE `([^`]*)`", sql_statement, re.IGNORECASE)
-    col_search = re.search("COLUMN `([^`]*)`", sql_statement, re.IGNORECASE)
-    return {
-        "err_msg": migration_test["err_msg"],
-        "code": migration_test["code"],
-        "table": table_search.group(1) if table_search else None,
-        "column": col_search.group(1) if col_search else None,
-    }
-
-
-def analyse_sql_statements(
-    sql_statements, database_vendor, exclude_migration_tests=None
-):
-    exclude_migration_tests = exclude_migration_tests or []
-    errors, ignored = [], []
-    for statement in sql_statements:
-        for test in get_single_line_migration_tests(database_vendor):
-            if test["fn"](statement, errors=errors):
-                if test["code"] in exclude_migration_tests:
-                    logger.debug("Testing {0} -- IGNORED".format(statement))
-                    err = build_error_dict(migration_test=test, sql_statement=statement)
-                    ignored.append(err)
-                else:
-                    logger.debug("Testing {0} -- ERROR".format(statement))
-                    err = build_error_dict(migration_test=test, sql_statement=statement)
-                    errors.append(err)
-            else:
-                logger.debug("Testing {0} -- PASSED".format(statement))
-
-    for test in get_transaction_migration_tests(database_vendor):
-        if test["fn"](sql_statements, errors=errors):
-            if test["code"] in exclude_migration_tests:
-                logger.debug("Testing {0} -- IGNORED".format(sql_statements))
-                err = build_error_dict(migration_test=test)
-                ignored.append(err)
-            else:
-                logger.debug("Testing {0} -- ERROR".format(sql_statements))
-                err = build_error_dict(migration_test=test)
-                errors.append(err)
-        else:
-            logger.debug("Testing {0} -- PASSED".format(sql_statements))
-
-    return errors, ignored

django_migration_linter/sql_analyser/__init__.py

@@ -0,0 +1,19 @@
+# Copyright 2019 3YOURMIND GmbH
+
+# Licensed under the Apache License, Version 2.0 (the "License");
+# you may not use this file except in compliance with the License.
+# You may obtain a copy of the License at
+
+# http://www.apache.org/licenses/LICENSE-2.0
+
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+
+from .base import BaseAnalyser  # noqa
+from .mysql import MySqlAnalyser  # noqa
+from .postgresql import PostgresqlAnalyser  # noqa
+from .sqlite import SqliteAnalyser  # noqa
+from .analyser import analyse_sql_statements  # noqa

django_migration_linter/sql_analyser/analyser.py

@@ -0,0 +1,46 @@
+# Copyright 2019 3YOURMIND GmbH
+
+# Licensed under the Apache License, Version 2.0 (the "License");
+# you may not use this file except in compliance with the License.
+# You may obtain a copy of the License at
+
+# http://www.apache.org/licenses/LICENSE-2.0
+
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+
+import logging
+
+from django_migration_linter.sql_analyser import (
+    BaseAnalyser,
+    MySqlAnalyser,
+    PostgresqlAnalyser,
+    SqliteAnalyser,
+)
+
+logger = logging.getLogger(__name__)
+
+
+def get_sql_analyser(database_vendor, exclude_migration_tests=None):
+    if "mysql" in database_vendor:
+        sql_analyser_class = MySqlAnalyser
+    elif "postgre" in database_vendor:
+        sql_analyser_class = PostgresqlAnalyser
+    elif "sqlite" in database_vendor:
+        sql_analyser_class = SqliteAnalyser
+    else:
+        sql_analyser_class = BaseAnalyser
+
+    logger.debug("Chosen SQL analyser class: %s", sql_analyser_class)
+    return sql_analyser_class(exclude_migration_tests)
+
+
+def analyse_sql_statements(
+    sql_statements, database_vendor, exclude_migration_tests=None
+):
+    sql_analyser = get_sql_analyser(database_vendor, exclude_migration_tests)
+    sql_analyser.analyse(sql_statements)
+    return sql_analyser.errors, sql_analyser.ignored

django_migration_linter/sql_analyser/base.py

@@ -0,0 +1,109 @@
+# Copyright 2019 3YOURMIND GmbH
+
+# Licensed under the Apache License, Version 2.0 (the "License");
+# you may not use this file except in compliance with the License.
+# You may obtain a copy of the License at
+
+# http://www.apache.org/licenses/LICENSE-2.0
+
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+
+import logging
+import re
+
+from .utils import update_migration_tests, build_error_dict
+
+logger = logging.getLogger(__name__)
+
+
+def has_not_null_column(sql_statements, **kwargs):
+    # TODO: improve to detect that the same column is concerned
+    return any(
+        re.search("(?<!DROP )NOT NULL", sql) and not sql.startswith("CREATE TABLE")
+        for sql in sql_statements
+    ) and not any("SET DEFAULT" in sql for sql in sql_statements)
+
+
+class BaseAnalyser(object):
+    base_migration_tests = [
+        {
+            "code": "RENAME_TABLE",
+            "fn": lambda sql, **kw: re.search("RENAME TABLE", sql)
+            or re.search("ALTER TABLE .* RENAME TO", sql),
+            "err_msg": "RENAMING tables",
+            "mode": "one_liner",
+        },
+        {
+            "code": "NOT_NULL",
+            "fn": has_not_null_column,
+            "err_msg": "NOT NULL constraint on columns",
+            "mode": "transaction",
+        },
+        {
+            "code": "DROP_COLUMN",
+            "fn": lambda sql, **kw: re.search("DROP COLUMN", sql),
+            "err_msg": "DROPPING columns",
+            "mode": "one_liner",
+        },
+        {
+            "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",
+            "mode": "one_liner",
+        },
+        {
+            "code": "ALTER_COLUMN",
+            "fn": lambda sql, **kw: re.search(
+                "ALTER TABLE .* ALTER COLUMN .* TYPE", sql
+            ),
+            "err_msg": (
+                "ALTERING columns (Could be backward compatible. "
+                "You may ignore this migration.)"
+            ),
+            "mode": "one_liner",
+        },
+    ]
+
+    migration_tests = []
+
+    def __init__(self, exclude_migration_tests):
+        self.exclude_migration_tests = exclude_migration_tests or []
+        self.errors = []
+        self.ignored = []
+        self.migration_tests = update_migration_tests(
+            self.base_migration_tests, self.migration_tests
+        )
+
+    def analyse(self, sql_statements):
+        for statement in sql_statements:
+            for test in self.one_line_migration_tests:
+                self._test_sql(test, sql=statement)
+
+        for test in self.transaction_migration_tests:
+            self._test_sql(test, sql=sql_statements)
+
+    @property
+    def one_line_migration_tests(self):
+        return [test for test in self.migration_tests if test["mode"] == "one_liner"]
+
+    @property
+    def transaction_migration_tests(self):
+        return [test for test in self.migration_tests if test["mode"] == "transaction"]
+
+    def _test_sql(self, test, sql):
+        if test["fn"](sql, errors=self.errors):
+            if test["code"] in self.exclude_migration_tests:
+                logger.debug("Testing %s -- IGNORED", sql)
+                err = build_error_dict(migration_test=test, sql_statement=sql)
+                self.ignored.append(err)
+            else:
+                logger.debug("Testing %s -- ERROR", sql)
+                err = build_error_dict(migration_test=test, sql_statement=sql)
+                self.errors.append(err)
+        else:
+            logger.debug("Testing %s -- PASSED", sql)

django_migration_linter/sql_analyser/mysql.py

@@ -0,0 +1,29 @@
+# Copyright 2019 3YOURMIND GmbH
+
+# Licensed under the Apache License, Version 2.0 (the "License");
+# you may not use this file except in compliance with the License.
+# You may obtain a copy of the License at
+
+# http://www.apache.org/licenses/LICENSE-2.0
+
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+
+import re
+
+from .base import BaseAnalyser
+
+
+class MySqlAnalyser(BaseAnalyser):
+    migration_tests = [
+        {
+            "code": "ALTER_COLUMN",
+            "fn": lambda sql, **kw: re.search(
+                "ALTER TABLE .* MODIFY .* (?!NULL);?$", sql
+            ),
+            "mode": "one_liner",
+        }
+    ]

django_migration_linter/sql_analyser/postgresql.py

@@ -0,0 +1,19 @@
+# Copyright 2019 3YOURMIND GmbH
+
+# Licensed under the Apache License, Version 2.0 (the "License");
+# you may not use this file except in compliance with the License.
+# You may obtain a copy of the License at
+
+# http://www.apache.org/licenses/LICENSE-2.0
+
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+
+from .base import BaseAnalyser
+
+
+class PostgresqlAnalyser(BaseAnalyser):
+    migration_tests = []

django_migration_linter/sql_analyser/sqlite.py

@@ -0,0 +1,41 @@
+# Copyright 2019 3YOURMIND GmbH
+
+# Licensed under the Apache License, Version 2.0 (the "License");
+# you may not use this file except in compliance with the License.
+# You may obtain a copy of the License at
+
+# http://www.apache.org/licenses/LICENSE-2.0
+
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+
+import re
+
+from .base import BaseAnalyser
+
+
+class SqliteAnalyser(BaseAnalyser):
+    migration_tests = [
+        {
+            "code": "RENAME_TABLE",
+            "fn": lambda sql, **kw: re.search("ALTER TABLE .* RENAME TO", sql)
+            and "__old" not in sql
+            and "new__" not in sql,
+        },
+        {
+            "code": "NOT_NULL",
+            "fn": lambda sql_statements, **kw: any(
+                re.search("NOT NULL(?! PRIMARY)(?! DEFAULT)", sql)
+                for sql in sql_statements
+            )
+            and any(
+                re.search("ALTER TABLE .* RENAME TO", sql)
+                and ("__old" in sql or "new__" in sql)
+                for sql in sql_statements
+            ),
+            "mode": "transaction",
+        },
+    ]

django_migration_linter/sql_analyser/utils.py

@@ -0,0 +1,55 @@
+# Copyright 2019 3YOURMIND GmbH
+
+# Licensed under the Apache License, Version 2.0 (the "License");
+# you may not use this file except in compliance with the License.
+# You may obtain a copy of the License at
+
+# http://www.apache.org/licenses/LICENSE-2.0
+
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+
+import re
+from copy import deepcopy
+
+
+def find_error_dict_with_code(tests, code):
+    return next((test_dict for test_dict in tests if test_dict["code"] == code), None)
+
+
+def update_migration_tests(base_tests, specific_tests):
+    base_tests = deepcopy(base_tests)
+    for override_test in specific_tests:
+        migration_test_dict = find_error_dict_with_code(
+            base_tests, override_test["code"]
+        )
+
+        if migration_test_dict is None or not override_test["code"]:
+            migration_test_dict = {}
+            base_tests.append(migration_test_dict)
+
+        for key in override_test.keys():
+            migration_test_dict[key] = override_test[key]
+    return base_tests
+
+
+def build_error_dict(migration_test, sql_statement):
+    table_search = (
+        re.search("TABLE `([^`]*)`", sql_statement, re.IGNORECASE)
+        if isinstance(sql_statement, str)
+        else None
+    )
+    col_search = (
+        re.search("COLUMN `([^`]*)`", sql_statement, re.IGNORECASE)
+        if isinstance(sql_statement, str)
+        else None
+    )
+    return {
+        "err_msg": migration_test["err_msg"],
+        "code": migration_test["code"],
+        "table": table_search.group(1) if table_search else None,
+        "column": col_search.group(1) if col_search else None,
+    }