codekingpro/portable-devtools

1import sqlparse2 3 4def query_starts_with(formatted_sql, prefixes):5    """Check if the query starts with any item from *prefixes*."""6    prefixes = [prefix.lower() for prefix in prefixes]7    return bool(formatted_sql) and formatted_sql.split()[0] in prefixes8 9 10def query_is_unconditional_update(formatted_sql):11    """Check if the query starts with UPDATE and contains no WHERE."""12    tokens = formatted_sql.split()13    return bool(tokens) and tokens[0] == "update" and "where" not in tokens14 15 16def query_is_simple_update(formatted_sql):17    """Check if the query starts with UPDATE."""18    tokens = formatted_sql.split()19    return bool(tokens) and tokens[0] == "update"20 21 22def is_destructive(queries, warning_level="all"):23    """Returns if any of the queries in *queries* is destructive."""24    keywords = ("drop", "shutdown", "delete", "truncate", "alter")25    for query in sqlparse.split(queries):26        if query:27            formatted_sql = sqlparse.format(query.lower(),28                                            strip_comments=True).strip()29            if query_starts_with(formatted_sql, keywords):30                return True31            if query_is_unconditional_update(formatted_sql):32                return True33            if warning_level == "all" and \34                    query_is_simple_update(formatted_sql):35                return True36    return False37