Get names of tables in the database
(username=get_current_user())
| 177 | |
| 178 | |
| 179 | def get_table_names(username=get_current_user()) -> List[str]: |
| 180 | """ |
| 181 | Get names of tables in the database |
| 182 | """ |
| 183 | IGNORE_TABLES = ["ai_sql_table_metadata", "ai_sql_type_metadata", "ai_sql_in_context_examples"] |
| 184 | try: |
| 185 | with ENGINE.connect() as connection: |
| 186 | connection = connection.execution_options( |
| 187 | postgresql_readonly=True |
| 188 | ) |
| 189 | with connection.begin(): |
| 190 | # sql_text = text(f""" |
| 191 | # SELECT tablename |
| 192 | # FROM pg_catalog.pg_tables |
| 193 | # WHERE tableowner = '{username}'; |
| 194 | # """) |
| 195 | sql_text = text(f""" |
| 196 | SELECT tablename |
| 197 | FROM pg_catalog.pg_tables |
| 198 | WHERE schemaname = 'public'; |
| 199 | """) |
| 200 | result = connection.execute(sql_text) |
| 201 | rows = [list(r) for r in result.all()] |
| 202 | |
| 203 | table_names = [] |
| 204 | for row in rows: |
| 205 | if row[0] not in IGNORE_TABLES: |
| 206 | table_names.append(row[0]) |
| 207 | |
| 208 | return table_names |
| 209 | |
| 210 | except Exception as e: |
| 211 | print(e) |
| 212 | return None |
| 213 | |
| 214 | |
| 215 | def generate_type_metadata(type_name): |
no test coverage detected