Spaces:
Sleeping
Sleeping
| from typing import Generator, Annotated | |
| from sqlmodel import Session, select | |
| from fastapi import Depends, HTTPException, status | |
| from ..core.database import engine | |
| from ..core.security import oauth2_scheme, decode_token | |
| from ..models.user import User | |
| def get_session() -> Generator[Session, None, None]: | |
| with Session(engine) as session: | |
| yield session | |
| SessionDep = Annotated[Session, Depends(get_session)] | |
| def get_current_user( | |
| session: SessionDep, | |
| token: Annotated[str, Depends(oauth2_scheme)] | |
| ) -> User: | |
| user_id = decode_token(token) | |
| if not user_id: | |
| raise HTTPException( | |
| status_code=status.HTTP_401_UNAUTHORIZED, | |
| detail="Could not validate credentials", | |
| headers={"WWW-Authenticate": "Bearer"}, | |
| ) | |
| user = session.get(User, int(user_id)) | |
| if not user: | |
| raise HTTPException( | |
| status_code=status.HTTP_404_NOT_FOUND, | |
| detail="User not found" | |
| ) | |
| return user | |
| CurrentUserDep = Annotated[User, Depends(get_current_user)] | |