Spaces:
Sleeping
Sleeping
File size: 1,043 Bytes
a291087 | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 | 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)]
|