Get sample data (3 rows) for all tables to help AI understand the data
(session: SessionDep, current_user: CurrentUser, ds: CoreDatasource,
table_list: list[str] = None)
| 488 | |
| 489 | |
| 490 | def get_tables_sample_data(session: SessionDep, current_user: CurrentUser, ds: CoreDatasource, |
| 491 | table_list: list[str] = None) -> str: |
| 492 | """Get sample data (3 rows) for all tables to help AI understand the data""" |
| 493 | table_objs = get_table_obj_by_ds(session=session, current_user=current_user, ds=ds) |
| 494 | if len(table_objs) == 0: |
| 495 | return "" |
| 496 | |
| 497 | sample_data_parts = [] |
| 498 | for obj in table_objs: |
| 499 | if table_list is not None and obj.table.table_name not in table_list: |
| 500 | continue |
| 501 | if obj.fields: |
| 502 | sample = get_table_sample_data(ds, obj.table.table_name, obj.fields) |
| 503 | if sample: |
| 504 | sample_data_parts.append(f"# Table: {obj.table.table_name}\n{sample}") |
| 505 | return "\n".join(sample_data_parts) |
| 506 | |
| 507 | |
| 508 | def get_table_schema(session: SessionDep, current_user: CurrentUser, ds: CoreDatasource, question: str, |
no test coverage detected