(session: SessionDep, token: TokenDep)
| 24 | |
| 25 | |
| 26 | def get_current_user(session: SessionDep, token: TokenDep) -> User: |
| 27 | try: |
| 28 | payload = jwt.decode(token, settings.SECRET_KEY, algorithms=[ALGORITHM]) |
| 29 | token_data = TokenPayload(**payload) |
| 30 | except (InvalidTokenError, ValidationError): |
| 31 | raise HTTPException( |
| 32 | status_code=status.HTTP_403_FORBIDDEN, |
| 33 | detail="Could not validate credentials", |
| 34 | ) |
| 35 | user = session.get(User, token_data.sub) |
| 36 | if not user: |
| 37 | raise HTTPException(status_code=404, detail="User not found") |
| 38 | if not user.is_active: |
| 39 | raise HTTPException(status_code=400, detail="Inactive user") |
| 40 | return user |
| 41 | |
| 42 | |
| 43 | CurrentUser = Annotated[User, Depends(get_current_user)] |
nothing calls this directly
no test coverage detected