在之前的分享的中,产生了token,用于做登录的认证,那么产生了token后,登陆携带了token,我们如何校验呢?
我们先来梳理下逻辑。
.调用登录接口,产生token .调用依赖登录的接口,在请求头中携带token .依赖登录的接口,接受到请求,判断是否在headers中携带token .携带token,校验是否过期 .解析token,获取username .根据username从redis获取, .查询到username的token且token相等,我们任务用户登录 .调用接口返回数据 .如果有其中一项校验不通过,返回对应的失败的信息
上面,我们梳理了思路,接下来我们就是去实现了。
我们先来实现从heders中获取token并且校验是否在redis有这个token。
async def get_cure_user(request: Request, token: Optional[str] = Header(...)) -> UserBase: #token依赖请求头的token校验 #验证失败返回信息 credentials_exception = HTTPException( status_code=status.HTTP_401_UNAUTHORIZED, detail="验证失败" ) #未登录失效的信息 credentials_FOR_exception = HTTPException( status_code=status.HTTP_403_FORBIDDEN, detail="用户未登录或者登陆token已经失效" ) try: #解析token payload = jwt.decode(token, SECRET_KEY, algorithms=[ALGORITHM]) #获取username username: str = payload.get("sub") #没有解析出来username if username is None: raise credentials_exception #从redis获取用户token useris = await request.app.state.redis.get(username) #token存在且存储的token一致认为成功 if not useris and useris!=token: raise credentials_FOR_exception #返回用户信息 user = UserBase(email=username) return user except JWTError: raise credentials_exception
UserBase类如下,存放在schemas中
from pydantic import BaseModel class UserBase(BaseModel): email: str
那么我们下面之前获取用户信息的时候,需要依赖登录,返回当前用户信息
@usersRouter.get("/user/getcuruser", response_model=Users) def getuser(users: UserBase = Depends(get_cure_user), db: Session = Depends(get_db)): user = get_user_emai(db, users.email) return user
那么我们去测试下,我们的设想是否实现。
没有携带token
那么我们调用下登录,
携带了token后就返回了当前用户的信息。这样我们可以把这个登录的认证,带入到其他的需要登录才能请求的接口中。
代码存储https://gitee.com/liwanlei/fastapistuday