51 lines
1.6 KiB
Python
51 lines
1.6 KiB
Python
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,
|
|
"type": "error",
|
|
},
|
|
{
|
|
"code": "DROP_TABLE",
|
|
# TODO: improve to detect that the table names overlap
|
|
"fn": lambda sql_statements, **kw: any(
|
|
sql.startswith("DROP TABLE") for sql in sql_statements
|
|
)
|
|
and not any(sql.startswith("CREATE TABLE") for sql in sql_statements),
|
|
"msg": "DROPPING table",
|
|
"mode": "transaction",
|
|
"type": "error",
|
|
},
|
|
{
|
|
"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",
|
|
"type": "error",
|
|
},
|
|
]
|
|
|
|
@staticmethod
|
|
def detect_table(sql):
|
|
if isinstance(sql, str):
|
|
regex_result = re.search("TABLE [`\"'](.*?)[`\"']", sql, re.IGNORECASE)
|
|
if regex_result:
|
|
return regex_result.group(1)
|
|
regex_result = re.search("ON [`\"'](.*?)[`\"']", sql, re.IGNORECASE)
|
|
if regex_result:
|
|
return regex_result.group(1)
|