(session: Session, llm_service)
| 34 | |
| 35 | |
| 36 | def get_assistant_ds(session: Session, llm_service) -> list[dict]: |
| 37 | assistant: AssistantHeader = llm_service.current_assistant |
| 38 | type = assistant.type |
| 39 | if type == 0 or type == 2: |
| 40 | configuration = assistant.configuration |
| 41 | if configuration: |
| 42 | config: dict[any] = json.loads(configuration) |
| 43 | oid: int = int(config['oid']) |
| 44 | stmt = select(CoreDatasource.id, CoreDatasource.name, CoreDatasource.description).where( |
| 45 | CoreDatasource.oid == oid) |
| 46 | if not assistant.online: |
| 47 | public_list: list[int] = config.get('public_list') or None |
| 48 | if public_list: |
| 49 | stmt = stmt.where(CoreDatasource.id.in_(public_list)) |
| 50 | else: |
| 51 | return [] |
| 52 | """ private_list: list[int] = config.get('private_list') or None |
| 53 | if private_list: |
| 54 | stmt = stmt.where(~CoreDatasource.id.in_(private_list)) """ |
| 55 | db_ds_list = session.exec(stmt) |
| 56 | |
| 57 | result_list = [ |
| 58 | { |
| 59 | "id": ds.id, |
| 60 | "name": ds.name, |
| 61 | "description": ds.description |
| 62 | } |
| 63 | for ds in db_ds_list |
| 64 | ] |
| 65 | |
| 66 | # filter private ds if offline |
| 67 | return result_list |
| 68 | out_ds_instance: AssistantOutDs = AssistantOutDsFactory.get_instance(assistant) |
| 69 | llm_service.out_ds_instance = out_ds_instance |
| 70 | dslist = out_ds_instance.get_simple_ds_list() |
| 71 | # format? |
| 72 | return dslist |
| 73 | |
| 74 | |
| 75 | def init_dynamic_cors(app: FastAPI): |
no test coverage detected