michaelliv
07/20/2023, 9:48 PMverify_session
I'd like to wrap it with my own extensions. Meaning, this is the behavior I'm looking for:
class SessionVerifier:
def __init__(self):
self._session_container: Optional[SessionContainer] = None
async def __call__(self):
if self._session_container is None:
self._session_container = verify_session()
return self._session_container
_session_verifier = SessionVerifier()
async def authenticate_user(
db_session: AsyncSession = Depends(get_db_session),
x_api_key: Optional[str] = Header(None),
session_verifier: SessionVerifier = Depends(_session_verifier),
):
if x_api_key:
is_verified = .... <verify jwt>
if is_verified:
# Somehow grab user_id
return UUID(user_id)
raise HTTPException(status_code=status.HTTP_401_UNAUTHORIZED, detail="Invalid API Key")
else:
user_session = await session_verifier()
if user_session is not None:
return UUID(user_session.get_user_id())
raise HTTPException(status_code=status.HTTP_401_UNAUTHORIZED, detail="Could not validate credentials")
Basically:
1) First authenticate with jwt token
2) If no api key is passed via header - try to authenticate with session
Then for my endpoints i'd use it like that:
@event_log_router.get("/events", response_model=EventLogsRead)
async def read_event_log(
user_id: UUID = Depends(authenticate_user),
db_session: AsyncSession = Depends(get_db_session),
):
pass
But obviously I'm missing how it should be used 🙂
Please advice, and thank you in advance!