(session: SessionDep, token: str)
| 50 | |
| 51 | |
| 52 | def get_user(session: SessionDep, token: str): |
| 53 | try: |
| 54 | payload = jwt.decode( |
| 55 | token, settings.SECRET_KEY, algorithms=[security.ALGORITHM] |
| 56 | ) |
| 57 | token_data = TokenPayload(**payload) |
| 58 | except (InvalidTokenError, ValidationError): |
| 59 | raise HTTPException( |
| 60 | status_code=status.HTTP_403_FORBIDDEN, |
| 61 | detail="Could not validate credentials", |
| 62 | ) |
| 63 | # session_user = await get_user_info(session=session, user_id=token_data.id) |
| 64 | |
| 65 | db_user: UserModel = get_db_user(session=session, user_id=token_data.id) |
| 66 | session_user = UserInfoDTO.model_validate(db_user.model_dump()) |
| 67 | session_user.isAdmin = session_user.id == 1 and session_user.account == 'admin' |
| 68 | session_user.language = 'zh-CN' |
| 69 | if session_user.isAdmin: |
| 70 | session_user = session_user |
| 71 | ws_model: UserWsModel = session.exec( |
| 72 | select(UserWsModel).where(UserWsModel.uid == session_user.id, UserWsModel.oid == session_user.oid)).first() |
| 73 | session_user.weight = ws_model.weight if ws_model else -1 |
| 74 | |
| 75 | session_user = UserInfoDTO.model_validate(session_user) |
| 76 | if not session_user: |
| 77 | raise HTTPException(status_code=404, detail="User not found") |
| 78 | |
| 79 | if session_user.status != 1: |
| 80 | raise HTTPException(status_code=400, detail="Inactive user") |
| 81 | return session_user |
| 82 | |
| 83 | |
| 84 | @router.post("/mcp_start", operation_id="mcp_start") |
no test coverage detected