Spaces:
Running
Running
| # read_me (route) | |
| # | | |
| # └── get_user_with_permissions | |
| # ├── get_database | |
| # └── get_current_user | |
| # └── HTTPBearer (security) | |
| from fastapi import Depends, HTTPException, status | |
| from fastapi.security import HTTPBearer, HTTPAuthorizationCredentials | |
| from jwt import decode, InvalidTokenError | |
| from ..core.config.config import settings | |
| from datetime import datetime | |
| async def get_database(): | |
| """ | |
| A placeholder function to simulate database access. | |
| In a real application, this would return a database session/connection. | |
| """ | |
| # For now, returns None - implement actual DB connection | |
| try: | |
| yield None | |
| finally: | |
| pass | |
| security = HTTPBearer() ## we will get the header of the http request by using this method | |
| async def get_current_user(credentials: HTTPAuthorizationCredentials = Depends(security)): | |
| """ | |
| A function to get the current user based on the provided token. | |
| this dependency: | |
| - extract the token from the request header | |
| - validate it | |
| - return the user information if valid. | |
| """ | |
| token = credentials.credentials # It returns the raw token value (without Bearer) from the request header. | |
| # Validate token | |
| try: | |
| payload = decode(token, settings.SECRET_KEY, algorithms=[settings.JWT_ALGORITHM]) | |
| user_id: str = payload.get("sub") | |
| if user_id is None: | |
| raise HTTPException( | |
| status_code=status.HTTP_401_UNAUTHORIZED, | |
| detail="Invalid token" | |
| ) | |
| except InvalidTokenError: | |
| raise HTTPException( | |
| status_code=status.HTTP_401_UNAUTHORIZED, | |
| detail="Invalid or expired token" | |
| ) | |
| return {"user_id": user_id} | |
| async def get_user_with_permissions( | |
| db = Depends(get_database), | |
| user = Depends(get_current_user) | |
| ): | |
| # permissions = await fetch_user_permissions(db, user.id) | |
| # user.permissions = permissions | |
| return user | |