| 101 | |
| 102 | |
| 103 | class ParsedQuery: |
| 104 | def __init__(self, sql_statement: str, strip_comments: bool = False): |
| 105 | if strip_comments: |
| 106 | sql_statement = sqlparse.format(sql_statement, strip_comments=True) |
| 107 | |
| 108 | self.sql: str = sql_statement |
| 109 | self._tables: Set[Table] = set() |
| 110 | self._alias_names: Set[str] = set() |
| 111 | self._limit: Optional[int] = None |
| 112 | |
| 113 | logger.debug("Parsing with sqlparse statement: %s", self.sql) |
| 114 | self._parsed = sqlparse.parse(self.stripped()) |
| 115 | for statement in self._parsed: |
| 116 | self._limit = _extract_limit_from_query(statement) |
| 117 | |
| 118 | @property |
| 119 | def tables(self) -> Set[Table]: |
| 120 | if not self._tables: |
| 121 | for statement in self._parsed: |
| 122 | self._extract_from_token(statement) |
| 123 | |
| 124 | self._tables = { |
| 125 | table for table in self._tables if str(table) not in self._alias_names |
| 126 | } |
| 127 | return self._tables |
| 128 | |
| 129 | @property |
| 130 | def limit(self) -> Optional[int]: |
| 131 | return self._limit |
| 132 | |
| 133 | def is_select(self) -> bool: |
| 134 | # make sure we strip comments; prevents a bug with coments in the CTE |
| 135 | parsed = sqlparse.parse(self.strip_comments()) |
| 136 | return parsed[0].get_type() == "SELECT" |
| 137 | |
| 138 | def is_valid_ctas(self) -> bool: |
| 139 | parsed = sqlparse.parse(self.strip_comments()) |
| 140 | return parsed[-1].get_type() == "SELECT" |
| 141 | |
| 142 | def is_valid_cvas(self) -> bool: |
| 143 | parsed = sqlparse.parse(self.strip_comments()) |
| 144 | return len(parsed) == 1 and parsed[0].get_type() == "SELECT" |
| 145 | |
| 146 | def is_explain(self) -> bool: |
| 147 | # Remove comments |
| 148 | statements_without_comments = sqlparse.format( |
| 149 | self.stripped(), strip_comments=True |
| 150 | ) |
| 151 | |
| 152 | # Explain statements will only be the first statement |
| 153 | return statements_without_comments.startswith("EXPLAIN") |
| 154 | |
| 155 | def is_show(self) -> bool: |
| 156 | # Remove comments |
| 157 | statements_without_comments = sqlparse.format( |
| 158 | self.stripped(), strip_comments=True |
| 159 | ) |
| 160 | # Show statements will only be the first statement |
no outgoing calls