在之前的分享的中,產生了token,用於做登錄的認證,那么產生了token后,登陸攜帶了token,我們如何校驗呢?
我們先來梳理下邏輯。
1.調用登錄接口,產生token 2.調用依賴登錄的接口,在請求頭中攜帶token 3.依賴登錄的接口,接受到請求,判斷是否在headers中攜帶token 4.攜帶token,校驗是否過期 5.解析token,獲取username 6.根據username從redis獲取, 7.查詢到username的token且token相等,我們任務用戶登錄 8.調用接口返回數據 9.如果有其中一項校驗不通過,返回對應的失敗的信息
上面,我們梳理了思路,接下來我們就是去實現了。
我們先來實現從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
文章首發在公眾號,歡迎關注。