auth.py 1.2 KB

12345678910111213141516171819202122232425262728293031323334353637383940414243
  1. from fastapi import HTTPException, Query, Depends
  2. from fastapi.security import OAuth2PasswordBearer
  3. from typing import Optional, Annotated
  4. from pydantic import BaseModel
  5. import jwt
  6. from ..db.redis_client import get_app_user
  7. from ..config.config import Config
  8. config = Config()
  9. oauth2_scheme_optional = OAuth2PasswordBearer(tokenUrl="/users/token", auto_error=False)
  10. class UserContext(BaseModel):
  11. user_id: str
  12. app_name: str
  13. async def resolve_user_id(
  14. jwt_token: Annotated[Optional[str], Depends(oauth2_scheme_optional)] = None,
  15. ) -> UserContext:
  16. if not jwt_token:
  17. raise HTTPException(status_code=401, detail="未提供认证令牌")
  18. try:
  19. payload = jwt.decode(jwt_token, config.SECRET_KEY, algorithms=[config.ALGORITHM])
  20. appName=payload.get("appName")
  21. user_id = payload.get("userId")
  22. if not user_id:
  23. raise HTTPException(status_code=401, detail="无效的令牌")
  24. if not appName:
  25. appName = "com.yunxiangshengtai"
  26. return UserContext(
  27. user_id=user_id,
  28. app_name=appName,
  29. )
  30. except jwt.PyJWTError:
  31. raise HTTPException(status_code=401, detail="无效的令牌")