(session: SessionDep, current_assistant: CurrentAssistant)
| 168 | |
| 169 | @router.get("/ds", include_in_schema=False, response_model=list[dict]) |
| 170 | async def ds(session: SessionDep, current_assistant: CurrentAssistant): |
| 171 | if current_assistant.type == 0: |
| 172 | online = current_assistant.online |
| 173 | configuration = current_assistant.configuration |
| 174 | config: dict[any] = json.loads(configuration) |
| 175 | oid: int = int(config['oid']) |
| 176 | stmt = select(CoreDatasource.id, CoreDatasource.name, CoreDatasource.description, CoreDatasource.type, CoreDatasource.type_name, CoreDatasource.num).where( |
| 177 | CoreDatasource.oid == oid) |
| 178 | if not online: |
| 179 | public_list: list[int] = config.get('public_list') or None |
| 180 | if public_list: |
| 181 | stmt = stmt.where(CoreDatasource.id.in_(public_list)) |
| 182 | else: |
| 183 | return [] |
| 184 | db_ds_list = session.exec(stmt) |
| 185 | return [ |
| 186 | { |
| 187 | "id": ds.id, |
| 188 | "name": ds.name, |
| 189 | "description": ds.description, |
| 190 | "type": ds.type, |
| 191 | "type_name": ds.type_name, |
| 192 | "num": ds.num, |
| 193 | } |
| 194 | for ds in db_ds_list] |
| 195 | if current_assistant.type == 1: |
| 196 | out_ds_instance: AssistantOutDs = AssistantOutDsFactory.get_instance(current_assistant) |
| 197 | return [ |
| 198 | { |
| 199 | "id": str(ds.id), |
| 200 | "name": ds.name, |
| 201 | "description": ds.description or ds.comment, |
| 202 | "type": ds.type, |
| 203 | "type_name": get_db_type(ds.type), |
| 204 | "num": len(ds.tables) if ds.tables else 0, |
| 205 | } |
| 206 | for ds in out_ds_instance.ds_list |
| 207 | if get_db_type(ds.type) |
| 208 | ] |
| 209 | |
| 210 | return None |
| 211 | |
| 212 | def get_db_type(type): |
| 213 | try: |
nothing calls this directly
no test coverage detected