Spaces:
Sleeping
Sleeping
| """ | |
| FastAPI Dependencies | |
| Authentication and authorization dependencies | |
| """ | |
| from fastapi import Depends, HTTPException, status | |
| from fastapi.security import HTTPBearer, HTTPAuthorizationCredentials | |
| from typing import Optional | |
| from app.utils.security import decode_access_token | |
| from app.db.repositories.user_repository import UserRepository | |
| from app.models.user import TokenData | |
| # HTTP Bearer token scheme | |
| security = HTTPBearer() | |
| async def get_current_user( | |
| credentials: HTTPAuthorizationCredentials = Depends(security) | |
| ) -> TokenData: | |
| """ | |
| Get current authenticated user from JWT token. | |
| This dependency extracts and validates the JWT token from the | |
| Authorization header and returns the user data. | |
| Args: | |
| credentials: HTTP Bearer credentials | |
| Returns: | |
| TokenData: User data from token | |
| Raises: | |
| HTTPException: If token is invalid or expired | |
| """ | |
| credentials_exception = HTTPException( | |
| status_code=status.HTTP_401_UNAUTHORIZED, | |
| detail="Could not validate credentials", | |
| headers={"WWW-Authenticate": "Bearer"}, | |
| ) | |
| # Decode token | |
| token = credentials.credentials | |
| payload = decode_access_token(token) | |
| if payload is None: | |
| raise credentials_exception | |
| # Extract user data | |
| user_id: str = payload.get("user_id") | |
| email: str = payload.get("email") | |
| if user_id is None or email is None: | |
| raise credentials_exception | |
| # Verify user exists | |
| user_repo = UserRepository() | |
| user = await user_repo.get_user_by_id(user_id) | |
| if user is None or not user.get("is_active", False): | |
| raise credentials_exception | |
| return TokenData(user_id=user_id, email=email) | |
| async def get_optional_current_user( | |
| credentials: Optional[HTTPAuthorizationCredentials] = Depends(security) | |
| ) -> Optional[TokenData]: | |
| """ | |
| Get current user if authenticated, None otherwise. | |
| This is a non-required version of get_current_user for optional auth. | |
| Args: | |
| credentials: Optional HTTP Bearer credentials | |
| Returns: | |
| TokenData or None: User data from token or None | |
| """ | |
| if credentials is None: | |
| return None | |
| try: | |
| return await get_current_user(credentials) | |
| except HTTPException: | |
| return None | |