(token: str = Depends(oauth2_scheme))
| 83 | |
| 84 | # Define the dependency function to get the current user |
| 85 | async def get_current_user(token: str = Depends(oauth2_scheme)) -> User: |
| 86 | credentials_exception = HTTPException( |
| 87 | status_code=status.HTTP_401_UNAUTHORIZED, |
| 88 | detail="Could not validate credentials", |
| 89 | headers={"WWW-Authenticate": "Bearer"}, |
| 90 | ) |
| 91 | try: |
| 92 | payload = jwt.decode(token, AppConfig.SECRET_KEY, algorithms=[AppConfig.ALGORITHM]) |
| 93 | username: str = payload.get("sub") |
| 94 | if username is None: |
| 95 | raise credentials_exception |
| 96 | # Here you should get the user from your Neo4j database using the username |
| 97 | # and convert it to the UserInDB model. |
| 98 | user = get_user_from_db(username) |
| 99 | if user is None: |
| 100 | raise credentials_exception |
| 101 | except JWTError: |
| 102 | raise credentials_exception |
| 103 | return user |
| 104 | |
| 105 | ## Endpoints |
| 106 | router = APIRouter() |
nothing calls this directly
no test coverage detected