Extract the column names from a select SQL statement. Returns a list of columns.
(sql: str)
| 219 | |
| 220 | |
| 221 | def extract_columns_from_select(sql: str) -> list[str]: |
| 222 | """ |
| 223 | Extract the column names from a select SQL statement. |
| 224 | |
| 225 | Returns a list of columns. |
| 226 | """ |
| 227 | parsed = sqlparse.parse(sql) |
| 228 | if not parsed: |
| 229 | return [] |
| 230 | |
| 231 | statement = get_last_select(parsed[0]) |
| 232 | |
| 233 | # if there is no select, skip checking for columns |
| 234 | if not statement: |
| 235 | return [] |
| 236 | |
| 237 | columns = [] |
| 238 | |
| 239 | # Loops through the tokens (pieces) of the SQL statement. |
| 240 | # Once it finds the SELECT token (generally first), it |
| 241 | # will then start looking for columns from that point on. |
| 242 | # The get_real_name() function returns the real column name |
| 243 | # even if an alias is used. |
| 244 | found_select = False |
| 245 | for token in statement.tokens: |
| 246 | if token.ttype is DML and token.value.upper() == 'SELECT': |
| 247 | found_select = True |
| 248 | elif found_select: |
| 249 | if isinstance(token, IdentifierList): |
| 250 | # multiple columns |
| 251 | for identifier in token.get_identifiers(): |
| 252 | if isinstance(identifier, Identifier): |
| 253 | column = identifier.get_real_name() |
| 254 | elif isinstance(identifier, Token): |
| 255 | column = identifier.value |
| 256 | else: |
| 257 | continue |
| 258 | columns.append(column) |
| 259 | elif isinstance(token, Identifier): |
| 260 | # single column |
| 261 | column = token.get_real_name() |
| 262 | columns.append(column) |
| 263 | elif token.ttype is Keyword: |
| 264 | break |
| 265 | |
| 266 | if columns: |
| 267 | break |
| 268 | return columns |
| 269 | |
| 270 | |
| 271 | def extract_tables_from_complete_statements(sql: str) -> list[tuple[str | None, str, str | None]]: |