(self, assistantToken: Optional[str], trans: I18n)
| 148 | |
| 149 | |
| 150 | async def validateAssistant(self, assistantToken: Optional[str], trans: I18n) -> tuple[any]: |
| 151 | if not assistantToken: |
| 152 | return False, f"Miss Token[{settings.TOKEN_KEY}]!" |
| 153 | schema, param = get_authorization_scheme_param(assistantToken) |
| 154 | |
| 155 | |
| 156 | try: |
| 157 | if schema.lower() == 'embedded': |
| 158 | return await self.validateEmbedded(param, trans) |
| 159 | if schema.lower() != "assistant": |
| 160 | return False, f"Token schema error!" |
| 161 | payload = jwt.decode( |
| 162 | param, settings.SECRET_KEY, algorithms=[security.ALGORITHM] |
| 163 | ) |
| 164 | token_data = TokenPayload(**payload) |
| 165 | if not payload['assistant_id']: |
| 166 | return False, f"Miss assistant payload error!" |
| 167 | with Session(engine) as session: |
| 168 | """ session_user = await get_user_info(session = session, user_id = token_data.id) |
| 169 | session_user = UserInfoDTO.model_validate(session_user) """ |
| 170 | session_user = get_assistant_user(id = token_data.id) |
| 171 | assistant_info = await get_assistant_info(session=session, assistant_id=payload['assistant_id']) |
| 172 | assistant_info = AssistantModel.model_validate(assistant_info) |
| 173 | assistant_info = AssistantHeader.model_validate(assistant_info.model_dump(exclude_unset=True)) |
| 174 | session_user.oid = int(assistant_info.oid) |
| 175 | |
| 176 | return True, session_user, assistant_info |
| 177 | except Exception as e: |
| 178 | SQLBotLogUtil.exception(f"Assistant validation error: {str(e)}") |
| 179 | # Return False and the exception message |
| 180 | return False, e |
| 181 | |
| 182 | async def validateEmbedded(self, param: str, trans: I18n) -> tuple[any]: |
| 183 | try: |
no test coverage detected