(session: SessionDep, id: int, virtual: Optional[int] = Query(None))
| 69 | |
| 70 | @router.get("/validator", response_model=AssistantValidator, include_in_schema=False) |
| 71 | async def validator(session: SessionDep, id: int, virtual: Optional[int] = Query(None)): |
| 72 | if not id: |
| 73 | raise Exception('miss assistant id') |
| 74 | |
| 75 | db_model = await get_assistant_info(session=session, assistant_id=id) |
| 76 | if not db_model: |
| 77 | return AssistantValidator() |
| 78 | db_model = AssistantModel.model_validate(db_model) |
| 79 | assistant_oid = 1 |
| 80 | if (db_model.type == 0): |
| 81 | configuration = db_model.configuration |
| 82 | config_obj = json.loads(configuration) if configuration else {} |
| 83 | assistant_oid = config_obj.get('oid', 1) |
| 84 | |
| 85 | access_token_expires = timedelta(minutes=settings.ACCESS_TOKEN_EXPIRE_MINUTES) |
| 86 | assistantDict = { |
| 87 | "id": virtual, "account": 'sqlbot-inner-assistant', "oid": assistant_oid, "assistant_id": id |
| 88 | } |
| 89 | access_token = create_access_token( |
| 90 | assistantDict, expires_delta=access_token_expires |
| 91 | ) |
| 92 | return AssistantValidator(True, True, True, access_token) |
| 93 | |
| 94 | |
| 95 | @router.get('/picture/{file_id}', summary=f"{PLACEHOLDER_PREFIX}assistant_picture_api", description=f"{PLACEHOLDER_PREFIX}assistant_picture_api") |
nothing calls this directly
no test coverage detected