Spaces:
Running
Running
| """Dependency injection for database sessions and JWT authentication. | |
| [Task]: T013, T014 | |
| [From]: specs/001-user-auth/quickstart.md | |
| """ | |
| import uuid | |
| from typing import Annotated, Optional | |
| from sqlmodel import Session | |
| from fastapi import Depends, HTTPException, status | |
| from starlette.requests import Request as StarletteRequest | |
| from fastapi.security import HTTPBearer, HTTPAuthorizationCredentials | |
| from core.database import get_session as db_get_session | |
| from core.security import decode_access_token | |
| # HTTP Bearer scheme for Authorization header | |
| security = HTTPBearer(auto_error=False) | |
| def get_session(): | |
| """Yield a database session with automatic cleanup. | |
| Uses the get_session function from core.database for consistency. | |
| """ | |
| yield from db_get_session() | |
| # Type alias for dependency injection | |
| SessionDep = Annotated[Session, Depends(get_session)] | |
| async def get_current_user_id( | |
| credentials: Optional[HTTPAuthorizationCredentials] = Depends(security), | |
| request: StarletteRequest = None | |
| ) -> uuid.UUID: | |
| """Get current user ID from JWT token. | |
| Extracts JWT token from Authorization header or httpOnly cookie, | |
| verifies it, and returns user_id as UUID. | |
| Args: | |
| credentials: HTTP Bearer credentials from Authorization header | |
| request: Starlette request object to access cookies | |
| Returns: | |
| Current authenticated user's ID as UUID | |
| Raises: | |
| HTTPException: If token is invalid, expired, or missing | |
| """ | |
| # Extract token from Authorization header or cookie | |
| token = None | |
| # Try Authorization header first | |
| if credentials: | |
| token = credentials.credentials | |
| # If no token in header, try httpOnly cookie | |
| if not token and request: | |
| auth_token = request.cookies.get("auth_token") | |
| if auth_token: | |
| token = auth_token | |
| if not token: | |
| raise HTTPException( | |
| status_code=status.HTTP_401_UNAUTHORIZED, | |
| detail="Could not validate credentials", | |
| headers={"WWW-Authenticate": "Bearer"}, | |
| ) | |
| try: | |
| # Decode and verify token | |
| payload = decode_access_token(token) | |
| user_id_str = payload.get("sub") | |
| if not user_id_str: | |
| raise HTTPException( | |
| status_code=status.HTTP_401_UNAUTHORIZED, | |
| detail="Invalid token: user_id missing" | |
| ) | |
| # Convert string to UUID for database comparison | |
| user_id = uuid.UUID(user_id_str) | |
| return user_id | |
| except HTTPException: | |
| raise | |
| except ValueError: | |
| raise HTTPException( | |
| status_code=status.HTTP_401_UNAUTHORIZED, | |
| detail="Invalid token: malformed user_id" | |
| ) | |
| except Exception as e: | |
| raise HTTPException( | |
| status_code=status.HTTP_401_UNAUTHORIZED, | |
| detail="Could not validate credentials" | |
| ) | |
| # Type alias for JWT authentication dependency | |
| CurrentUserDep = Annotated[uuid.UUID, Depends(get_current_user_id)] | |