34 lines
1.0 KiB
Python
34 lines
1.0 KiB
Python
from typing import Annotated
|
|
|
|
from fastapi import HTTPException, Depends
|
|
from fastapi.security import OAuth2PasswordBearer
|
|
from sqlalchemy import select
|
|
from starlette import status
|
|
|
|
from backend import config
|
|
from database import User
|
|
from jose import jwt
|
|
|
|
from database.base import DatabaseDependency
|
|
|
|
oauth2_scheme = OAuth2PasswordBearer("")
|
|
ALGORITHM = "HS256"
|
|
|
|
|
|
def generate_jwt_token(user: User):
|
|
return jwt.encode({'sub': user.id}, settings.SECRET_KEY, algorithm=ALGORITHM)
|
|
|
|
|
|
def require_jwt_sub(token: Annotated[str, Depends(oauth2_scheme)]):
|
|
payload = jwt.decode(token, settings.SECRET_KEY, algorithms=[ALGORITHM])
|
|
user_id = payload.get("sub")
|
|
if not user_id:
|
|
raise HTTPException(status_code=status.HTTP_401_UNAUTHORIZED, detail='Invalid authorization credentials')
|
|
return payload
|
|
|
|
|
|
async def get_current_user(db_session: DatabaseDependency, user_id: Annotated[int, Depends(require_jwt_sub)]) -> User:
|
|
user = await db_session.scalar(select(User).where(User.id == user_id))
|
|
if user:
|
|
return user
|