(*, session: Session, user_id: int)
| 25 | |
| 26 | @cache(namespace=CacheNamespace.AUTH_INFO, cacheName=CacheName.USER_INFO, keyExpression="user_id") |
| 27 | async def get_user_info(*, session: Session, user_id: int) -> UserInfoDTO | None: |
| 28 | db_user: UserModel = get_db_user(session = session, user_id = user_id) |
| 29 | if not db_user: |
| 30 | return None |
| 31 | userInfo = UserInfoDTO.model_validate(db_user.model_dump()) |
| 32 | userInfo.isAdmin = userInfo.id == 1 and userInfo.account == 'admin' |
| 33 | if userInfo.isAdmin: |
| 34 | return userInfo |
| 35 | ws_model: UserWsModel = session.exec(select(UserWsModel).where(UserWsModel.uid == userInfo.id, UserWsModel.oid == userInfo.oid)).first() |
| 36 | userInfo.weight = ws_model.weight if ws_model else -1 |
| 37 | return userInfo |
| 38 | |
| 39 | def authenticate(*, session: Session, account: str, password: str) -> BaseUserDTO | None: |
| 40 | db_user = get_user_by_account(session=session, account=account) |
no test coverage detected