(session: SessionDep, trans: Trans, user: CurrentUser, create_ds: CreateDatasource)
| 68 | |
| 69 | @clear_cache(namespace=CacheNamespace.AUTH_INFO, cacheName=CacheName.DS_ID_LIST, keyExpression="user.oid") |
| 70 | async def create_ds(session: SessionDep, trans: Trans, user: CurrentUser, create_ds: CreateDatasource): |
| 71 | ds = CoreDatasource() |
| 72 | deepcopy_ignore_extra(create_ds, ds) |
| 73 | check_name(session, trans, user, ds) |
| 74 | ds.create_time = datetime.datetime.now() |
| 75 | # status = check_status(session, ds) |
| 76 | ds.create_by = user.id |
| 77 | ds.oid = user.oid if user.oid is not None else 1 |
| 78 | ds.status = "Success" |
| 79 | ds.type_name = DB.get_db(ds.type).db_name |
| 80 | record = CoreDatasource(**ds.model_dump()) |
| 81 | session.add(record) |
| 82 | session.flush() |
| 83 | session.refresh(record) |
| 84 | ds.id = record.id |
| 85 | session.commit() |
| 86 | |
| 87 | # save tables and fields |
| 88 | sync_table(session, ds, create_ds.tables) |
| 89 | updateNum(session, ds) |
| 90 | return ds |
| 91 | |
| 92 | |
| 93 | def chooseTables(session: SessionDep, trans: Trans, id: int, tables: List[CoreTable]): |
no test coverage detected