Coverage for fss\common\security\security.py: 81%
21 statements
« prev ^ index » next coverage.py v7.4.4, created at 2024-04-11 19:09 +0800
« prev ^ index » next coverage.py v7.4.4, created at 2024-04-11 19:09 +0800
1from typing import Callable
3from fastapi import Depends, HTTPException
4from fastapi.security import OAuth2PasswordBearer
5from jose import ExpiredSignatureError
6from multipart.exceptions import DecodeError
7from starlette import status
9from fss.common.config import configs
10from fss.common.schema.schema import CurrentUser
11from fss.common.util.security import get_payload
13oauth2_scheme = OAuth2PasswordBearer(tokenUrl=f"{configs.api_version}/user/login")
16def get_current_user() -> Callable[[], CurrentUser]:
17 """
18 Acquire current info through access_token
19 :return: CurrentUser instance
20 """
22 async def current_user(
23 access_token: str = Depends(oauth2_scheme),
24 ) -> CurrentUser:
25 try:
26 payload = await get_payload(access_token)
27 except ExpiredSignatureError:
28 raise HTTPException(
29 status_code=status.HTTP_403_FORBIDDEN,
30 detail="Your token has expired. Please log in again.",
31 )
32 except DecodeError:
33 raise HTTPException(
34 status_code=status.HTTP_403_FORBIDDEN,
35 detail="Error when decoding the token. Please check your request.",
36 )
37 user_id = payload["sub"]
39 return CurrentUser(user_id=user_id)
41 return current_user