Simplify a query by converting CTEs into table metadata objects
(full_text, text_before_cursor)
| 15 | |
| 16 | |
| 17 | def isolate_query_ctes(full_text, text_before_cursor): |
| 18 | """Simplify a query by converting CTEs into table metadata objects |
| 19 | """ |
| 20 | |
| 21 | if not full_text: |
| 22 | return full_text, text_before_cursor, tuple() |
| 23 | |
| 24 | ctes, _ = extract_ctes(full_text) |
| 25 | if not ctes: |
| 26 | return full_text, text_before_cursor, () |
| 27 | |
| 28 | current_position = len(text_before_cursor) |
| 29 | meta = [] |
| 30 | |
| 31 | for cte in ctes: |
| 32 | if cte.start < current_position < cte.stop: |
| 33 | # Currently editing a cte - treat its body as the current full_text |
| 34 | text_before_cursor = full_text[cte.start:current_position] |
| 35 | full_text = full_text[cte.start:cte.stop] |
| 36 | return full_text, text_before_cursor, meta |
| 37 | |
| 38 | # Append this cte to the list of available table metadata |
| 39 | cols = (ColumnMetadata(name, None, ()) for name in cte.columns) |
| 40 | meta.append(TableMetadata(cte.name, cols)) |
| 41 | |
| 42 | # Editing past the last cte (ie the main body of the query) |
| 43 | full_text = full_text[ctes[-1].stop:] |
| 44 | text_before_cursor = text_before_cursor[ctes[-1].stop:current_position] |
| 45 | |
| 46 | return full_text, text_before_cursor, tuple(meta) |
| 47 | |
| 48 | |
| 49 | def extract_ctes(sql): |
no test coverage detected