new refresh dep add refresh endpoint

This commit is contained in:
IluaAir
2025-09-21 14:59:26 +03:00
parent c3bfb9cb6a
commit 42b8c3a2c9
12 changed files with 125 additions and 66 deletions

View File

@@ -1,7 +1,7 @@
from typing import Annotated
from fastapi import Depends, HTTPException, Path
from fastapi.security import OAuth2PasswordBearer
from fastapi.security import HTTPBearer, OAuth2PasswordBearer
from jwt import InvalidTokenError
from src.api.dependacies.db_dep import sessionDep
@@ -11,37 +11,46 @@ from src.schemas.auth import TokenData
from src.services.tasks import TaskService
from src.services.users import UserService
http_bearer = HTTPBearer(auto_error=False)
oauth2_scheme = OAuth2PasswordBearer(tokenUrl=f"{settings.api.v1_login_url}/login")
AccessTokenDep = Annotated[str, Depends(oauth2_scheme)]
async def get_current_user(token: Annotated[str, Depends(oauth2_scheme)]):
async def get_current_user(
token: AccessTokenDep, verify_exp: bool = True, check_active: bool = False
):
credentials_exception = HTTPException(
status_code=401,
detail="Could not validate credentials",
headers={"WWW-Authenticate": "Bearer"},
)
try:
payload = AuthManager.decode_access_token(token=token)
payload = AuthManager.decode_access_token(token, verify_exp)
if payload is None:
raise credentials_exception
user = TokenData(**payload)
if check_active and not user.is_active:
raise HTTPException(status_code=400, detail="Inactive user")
except InvalidTokenError:
raise credentials_exception
return user
CurrentUser = Annotated[TokenData, Depends(get_current_user)]
async def get_current_user_basic(token: AccessTokenDep):
return await get_current_user(token, verify_exp=True, check_active=False)
def get_current_active_user(
current_user: CurrentUser,
):
if not current_user.is_active:
raise HTTPException(status_code=400, detail="Inactive user")
return current_user
async def get_current_active_user(token: AccessTokenDep):
return await get_current_user(token, verify_exp=True, check_active=True)
async def get_current_user_for_refresh(token: AccessTokenDep):
return await get_current_user(token, verify_exp=False, check_active=True)
CurrentUser = Annotated[TokenData, Depends(get_current_user_basic)]
ActiveUser = Annotated[TokenData, Depends(get_current_active_user)]
RefreshUser = Annotated[TokenData, Depends(get_current_user_for_refresh)]
async def get_admin_user(db: sessionDep, current_user: ActiveUser):