(ds: CoreDatasource | AssistantOutDsSchema)
| 288 | |
| 289 | |
| 290 | def get_version(ds: CoreDatasource | AssistantOutDsSchema): |
| 291 | version = '' |
| 292 | if isinstance(ds, CoreDatasource): |
| 293 | conf = DatasourceConf( |
| 294 | **json.loads(aes_decrypt(ds.configuration))) if not equals_ignore_case(ds.type, |
| 295 | "excel") else get_engine_config() |
| 296 | else: |
| 297 | conf = DatasourceConf(**json.loads(aes_decrypt(get_out_ds_conf(ds, 10)))) |
| 298 | # if isinstance(ds, AssistantOutDsSchema): |
| 299 | # conf = DatasourceConf() |
| 300 | # conf.host = ds.host |
| 301 | # conf.port = ds.port |
| 302 | # conf.username = ds.user |
| 303 | # conf.password = ds.password |
| 304 | # conf.database = ds.dataBase |
| 305 | # conf.dbSchema = ds.db_schema |
| 306 | # conf.timeout = 10 |
| 307 | db = DB.get_db(ds.type) |
| 308 | sql = get_version_sql(ds, conf) |
| 309 | if not sql: |
| 310 | return '' |
| 311 | try: |
| 312 | if db.connect_type == ConnectType.sqlalchemy: |
| 313 | with get_session(ds) as session: |
| 314 | with session.execute(text(sql)) as result: |
| 315 | res = result.fetchall() |
| 316 | version = res[0][0] |
| 317 | else: |
| 318 | extra_config_dict = get_extra_config(conf) |
| 319 | if equals_ignore_case(ds.type, 'dm'): |
| 320 | with dmPython.connect(user=conf.username, password=conf.password, server=conf.host, |
| 321 | port=conf.port) as conn, conn.cursor() as cursor: |
| 322 | cursor.execute(sql, timeout=10, **extra_config_dict) |
| 323 | res = cursor.fetchall() |
| 324 | version = res[0][0] |
| 325 | elif equals_ignore_case(ds.type, 'doris', 'starrocks'): |
| 326 | ssl_args = {'ssl': {'ssl_mode': 'REQUIRE'}} if conf.ssl else {} |
| 327 | with pymysql.connect(user=conf.username, passwd=conf.password, host=conf.host, |
| 328 | port=conf.port, db=conf.database, connect_timeout=10, |
| 329 | read_timeout=10, **extra_config_dict, **ssl_args) as conn, conn.cursor() as cursor: |
| 330 | cursor.execute(sql) |
| 331 | res = cursor.fetchall() |
| 332 | version = res[0][0] |
| 333 | elif equals_ignore_case(ds.type, 'redshift', 'es', 'hive'): |
| 334 | version = '' |
| 335 | except Exception as e: |
| 336 | print(e) |
| 337 | version = '' |
| 338 | return version.decode() if isinstance(version, bytes) else version |
| 339 | |
| 340 | |
| 341 | def get_schema(ds: CoreDatasource): |
no test coverage detected