fix: Add get_current_user_optional for public endpoints
Some checks failed
Deploy Fediversion / deploy (push) Failing after 1s
Some checks failed
Deploy Fediversion / deploy (push) Failing after 1s
This commit is contained in:
parent
fae5349f9c
commit
0bdb7ca8f6
1 changed files with 23 additions and 0 deletions
|
|
@ -59,3 +59,26 @@ async def get_current_superuser(current_user: User = Depends(get_current_user)):
|
|||
detail="The user doesn't have enough privileges"
|
||||
)
|
||||
return current_user
|
||||
|
||||
|
||||
# Optional OAuth scheme that doesn't require auth
|
||||
oauth2_scheme_optional = OAuth2PasswordBearer(tokenUrl="auth/token", auto_error=False)
|
||||
|
||||
async def get_current_user_optional(
|
||||
token: Optional[str] = Depends(oauth2_scheme_optional),
|
||||
session: Session = Depends(get_session)
|
||||
) -> Optional[User]:
|
||||
"""Get current user if authenticated, otherwise return None (for public endpoints)"""
|
||||
if not token:
|
||||
return None
|
||||
try:
|
||||
payload = jwt.decode(token, SECRET_KEY, algorithms=[ALGORITHM])
|
||||
email: str = payload.get("sub")
|
||||
if email is None:
|
||||
return None
|
||||
except JWTError:
|
||||
return None
|
||||
|
||||
user = session.exec(select(User).where(User.email == email)).first()
|
||||
return user
|
||||
|
||||
|
|
|
|||
Loading…
Add table
Reference in a new issue