(session: SessionDep, current_user: CurrentUser, ds: CoreDatasource)
| 401 | |
| 402 | |
| 403 | def get_table_obj_by_ds(session: SessionDep, current_user: CurrentUser, ds: CoreDatasource) -> List[TableAndFields]: |
| 404 | _list: List = [] |
| 405 | tables = session.query(CoreTable).filter( |
| 406 | and_(CoreTable.ds_id == ds.id, CoreTable.checked == True) |
| 407 | ).all() |
| 408 | conf = DatasourceConf(**json.loads(aes_decrypt(ds.configuration))) if ds.type != "excel" else get_engine_config() |
| 409 | schema = conf.dbSchema if conf.dbSchema is not None and conf.dbSchema != "" else conf.database |
| 410 | |
| 411 | # get all field |
| 412 | table_ids = [table.id for table in tables] |
| 413 | all_fields = session.query(CoreField).filter( |
| 414 | and_(CoreField.table_id.in_(table_ids), CoreField.checked == True)).all() |
| 415 | # build dict |
| 416 | fields_dict = {} |
| 417 | for field in all_fields: |
| 418 | if fields_dict.get(field.table_id): |
| 419 | fields_dict.get(field.table_id).append(field) |
| 420 | else: |
| 421 | fields_dict[field.table_id] = [field] |
| 422 | |
| 423 | contain_rules = session.query(DsRules).all() |
| 424 | for table in tables: |
| 425 | # fields = session.query(CoreField).filter(and_(CoreField.table_id == table.id, CoreField.checked == True)).all() |
| 426 | fields = fields_dict.get(table.id) |
| 427 | |
| 428 | # do column permissions, filter fields |
| 429 | fields = get_column_permission_fields(session=session, current_user=current_user, table=table, fields=fields, |
| 430 | contain_rules=contain_rules) |
| 431 | _list.append(TableAndFields(schema=schema, table=table, fields=fields)) |
| 432 | return _list |
| 433 | |
| 434 | |
| 435 | def get_table_sample_data(ds: CoreDatasource, table_name: str, fields: list) -> str: |
no test coverage detected