Takes a parsed sql statement and returns the last select query where applicable. The intended use case is for when giving table suggestions based on columns, where we only want to look at the columns from the most recent select. This works for a single select query, or one or more
(parsed: TokenList)
| 90 | |
| 91 | |
| 92 | def get_last_select(parsed: TokenList) -> TokenList: |
| 93 | """ |
| 94 | Takes a parsed sql statement and returns the last select query where applicable. |
| 95 | |
| 96 | The intended use case is for when giving table suggestions based on columns, where |
| 97 | we only want to look at the columns from the most recent select. This works for a single |
| 98 | select query, or one or more sub queries (the useful part). |
| 99 | |
| 100 | The custom logic is necessary because the typical sqlparse logic for things like finding |
| 101 | sub selects (i.e. is_subselect) only works on complete statements, such as: |
| 102 | |
| 103 | * select c1 from t1; |
| 104 | |
| 105 | However when suggesting tables based on columns, we only have partial select statements, i.e.: |
| 106 | |
| 107 | * select c1 |
| 108 | * select c1 from (select c2) |
| 109 | |
| 110 | So given the above, we must parse them ourselves as they are not viewed as complete statements. |
| 111 | |
| 112 | Returns a TokenList of the last select statement's tokens. |
| 113 | """ |
| 114 | select_indexes: list[int] = [] |
| 115 | |
| 116 | for token in parsed: |
| 117 | if token.match(DML, "select"): # match is case insensitive |
| 118 | select_indexes.append(parsed.token_index(token)) |
| 119 | |
| 120 | last_select = TokenList() |
| 121 | |
| 122 | if select_indexes: |
| 123 | last_select = TokenList(parsed[select_indexes[-1] :]) |
| 124 | |
| 125 | return last_select |
| 126 | |
| 127 | |
| 128 | def extract_from_part(parsed: TokenList, stop_at_punctuation: bool = True) -> Generator[Any, None, None]: |
no outgoing calls