(session: SessionDep, id: int)
| 121 | |
| 122 | |
| 123 | async def delete_ds(session: SessionDep, id: int): |
| 124 | term = session.exec(select(CoreDatasource).where(CoreDatasource.id == id)).first() |
| 125 | if term.type == "excel": |
| 126 | # drop all tables for current datasource |
| 127 | engine = get_engine_conn() |
| 128 | conf = DatasourceConf(**json.loads(aes_decrypt(term.configuration))) |
| 129 | with engine.connect() as conn: |
| 130 | for sheet in conf.sheets: |
| 131 | conn.execute(text(f'DROP TABLE IF EXISTS "{sheet["tableName"]}"')) |
| 132 | conn.commit() |
| 133 | |
| 134 | session.delete(term) |
| 135 | session.commit() |
| 136 | delete_table_by_ds_id(session, id) |
| 137 | delete_field_by_ds_id(session, id) |
| 138 | if term: |
| 139 | await clear_ws_ds_cache(term.oid) |
| 140 | return { |
| 141 | "message": f"Datasource with ID {id} deleted successfully." |
| 142 | } |
| 143 | |
| 144 | |
| 145 | def getTables(session: SessionDep, id: int): |
no test coverage detected