(ds: CoreDatasource, table_name: str = None)
| 450 | |
| 451 | |
| 452 | def get_fields(ds: CoreDatasource, table_name: str = None): |
| 453 | conf = DatasourceConf(**json.loads(aes_decrypt(ds.configuration))) if not equals_ignore_case(ds.type, |
| 454 | "excel") else get_engine_config() |
| 455 | db = DB.get_db(ds.type) |
| 456 | sql, p1, p2 = get_field_sql(ds, conf, table_name) |
| 457 | if db.connect_type == ConnectType.sqlalchemy: |
| 458 | with get_session(ds) as session: |
| 459 | with session.execute(text(sql), {"param1": p1, "param2": p2}) as result: |
| 460 | res = result.fetchall() |
| 461 | res_list = [ColumnSchema(*item) for item in res] |
| 462 | return res_list |
| 463 | else: |
| 464 | extra_config_dict = get_extra_config(conf) |
| 465 | if equals_ignore_case(ds.type, 'dm'): |
| 466 | with dmPython.connect(user=conf.username, password=conf.password, server=conf.host, |
| 467 | port=conf.port, **extra_config_dict) as conn, conn.cursor() as cursor: |
| 468 | cursor.execute(sql, {"param1": p1, "param2": p2}, timeout=conf.timeout) |
| 469 | res = cursor.fetchall() |
| 470 | res_list = [ColumnSchema(*item) for item in res] |
| 471 | return res_list |
| 472 | elif equals_ignore_case(ds.type, 'doris', 'starrocks'): |
| 473 | ssl_args = {'ssl': {'ssl_mode': 'REQUIRE'}} if conf.ssl else {} |
| 474 | with pymysql.connect(user=conf.username, passwd=conf.password, host=conf.host, |
| 475 | port=conf.port, db=conf.database, connect_timeout=conf.timeout, |
| 476 | read_timeout=conf.timeout, **extra_config_dict, |
| 477 | **ssl_args) as conn, conn.cursor() as cursor: |
| 478 | cursor.execute(sql, (p1, p2)) |
| 479 | res = cursor.fetchall() |
| 480 | res_list = [ColumnSchema(*item) for item in res] |
| 481 | return res_list |
| 482 | elif equals_ignore_case(ds.type, 'redshift'): |
| 483 | with redshift_connector.connect(host=conf.host, port=conf.port, database=conf.database, user=conf.username, |
| 484 | password=conf.password, |
| 485 | timeout=conf.timeout, **extra_config_dict) as conn, conn.cursor() as cursor: |
| 486 | cursor.execute(sql, (p1, p2)) |
| 487 | res = cursor.fetchall() |
| 488 | res_list = [ColumnSchema(*item) for item in res] |
| 489 | return res_list |
| 490 | elif equals_ignore_case(ds.type, 'kingbase'): |
| 491 | with psycopg2.connect(host=conf.host, port=conf.port, database=conf.database, user=conf.username, |
| 492 | password=conf.password, |
| 493 | options=f"-c statement_timeout={conf.timeout * 1000}", |
| 494 | **extra_config_dict) as conn, conn.cursor() as cursor: |
| 495 | cursor.execute(sql.format(p1, p2)) |
| 496 | res = cursor.fetchall() |
| 497 | res_list = [ColumnSchema(*item) for item in res] |
| 498 | return res_list |
| 499 | elif equals_ignore_case(ds.type, 'es'): |
| 500 | res = get_es_fields(conf, table_name) |
| 501 | res_list = [ColumnSchema(*item) for item in res] |
| 502 | return res_list |
| 503 | elif equals_ignore_case(ds.type, 'hive'): |
| 504 | with hive.connect(host=conf.host, port=conf.port, username=conf.username, |
| 505 | database=conf.database, **extra_config_dict) as conn, conn.cursor() as cursor: |
| 506 | cursor.execute(sql) |
| 507 | res = cursor.fetchall() |
| 508 | res_list = [ColumnSchema(*item) for item in res] |
| 509 | return res_list |
no test coverage detected