perf: Assistant token

backend/common/core/config.py

@@ -57,8 +57,9 @@ class Settings(BaseSettings):
     POSTGRES_PASSWORD: str = ""
     POSTGRES_DB: str = ""
     
-    TOKEN_KEY: str
-    DEFAULT_PWD: str
+    TOKEN_KEY: str =  "X-SQLBOT-TOKEN"
+    DEFAULT_PWD: str = "SQLBot@123456"
+    ASSISTANT_TOKEN_KEY: str = "X-SQLBOT-ASSISTANT-TOKEN"
 
     @computed_field  # type: ignore[prop-decorator]
     @property

backend/common/core/deps.py

@@ -1,51 +1,21 @@
 from typing import Annotated
 
-import jwt
-from fastapi import Depends, HTTPException, Request, status
-# from fastapi.security import OAuth2PasswordBearer
-from jwt.exceptions import InvalidTokenError
-from pydantic import ValidationError
+from fastapi import Depends, Request
 from sqlmodel import Session
-from apps.system.crud.user import get_user_info
 from apps.system.schemas.system_schema import UserInfoDTO
-from common.core.schemas import TokenPayload, XOAuth2PasswordBearer
-from common.core import security
-from common.core.config import settings
 from common.core.db import get_session
 from common.utils.locale import I18n
-reusable_oauth2 = XOAuth2PasswordBearer(
-    tokenUrl=f"{settings.API_V1_STR}/login/access-token"
-)
 
 
-    
-
 SessionDep = Annotated[Session, Depends(get_session)]
-TokenDep = Annotated[str, Depends(reusable_oauth2)]
 i18n = I18n()
 async def get_i18n(request: Request):
     return i18n(request)
 
 Trans = Annotated[I18n, Depends(get_i18n)]
-async def get_current_user(session: SessionDep, token: TokenDep) -> UserInfoDTO:
-    try:
-        payload = jwt.decode(
-            token, settings.SECRET_KEY, algorithms=[security.ALGORITHM]
-        )
-        token_data = TokenPayload(**payload)
-    except (InvalidTokenError, ValidationError):
-        raise HTTPException(
-            status_code=status.HTTP_403_FORBIDDEN,
-            detail="Could not validate credentials",
-        )
-    session_user = await get_user_info(session = session, user_id = token_data.id)
-    session_user = UserInfoDTO.model_validate(session_user)
-    if not session_user:
-        raise HTTPException(status_code=404, detail="User not found")
-    
-    if session_user.status != 1:
-        raise HTTPException(status_code=400, detail="Inactive user")
-    return session_user
+async def get_current_user(request: Request) -> UserInfoDTO:
+    return request.state.current_user
+
 CurrentUser = Annotated[UserInfoDTO, Depends(get_current_user)]
 
 

backend/common/core/schemas.py

@@ -18,8 +18,11 @@ class Token(SQLModel):
 class XOAuth2PasswordBearer(OAuth2PasswordBearer):
     async def __call__(self, request: Request) -> Optional[str]:
         authorization = request.headers.get(settings.TOKEN_KEY)
+        if request.headers.get(settings.ASSISTANT_TOKEN_KEY):
+            authorization = request.headers.get(settings.ASSISTANT_TOKEN_KEY)
         scheme, param = get_authorization_scheme_param(authorization)
-        if not authorization or scheme.lower() != "bearer":
+        
+        if not authorization or scheme.lower() not in  ["bearer", "assistant"]:
             if self.auto_error:
                 raise HTTPException(
                     status_code=HTTP_401_UNAUTHORIZED,