| 12345678910111213141516171819202122232425262728293031323334353637383940414243 |
- from fastapi import HTTPException, Query, Depends
- from fastapi.security import OAuth2PasswordBearer
- from typing import Optional, Annotated
- from pydantic import BaseModel
- import jwt
- from ..db.redis_client import get_app_user
- from ..config.config import Config
- config = Config()
- oauth2_scheme_optional = OAuth2PasswordBearer(tokenUrl="/users/token", auto_error=False)
- class UserContext(BaseModel):
- user_id: str
- app_name: str
- async def resolve_user_id(
- jwt_token: Annotated[Optional[str], Depends(oauth2_scheme_optional)] = None,
- ) -> UserContext:
- if not jwt_token:
- raise HTTPException(status_code=401, detail="未提供认证令牌")
- try:
- payload = jwt.decode(jwt_token, config.SECRET_KEY, algorithms=[config.ALGORITHM])
- appName=payload.get("appName")
- user_id = payload.get("userId")
- if not user_id:
- raise HTTPException(status_code=401, detail="无效的令牌")
- if not appName:
- appName = "com.yunxiangshengtai"
- return UserContext(
- user_id=user_id,
- app_name=appName,
- )
- except jwt.PyJWTError:
- raise HTTPException(status_code=401, detail="无效的令牌")
|