FastAPI 学习之路（五十七）校验是否登陆

2024-04-12 21:03:20

在之前的分享的中，产生了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

文章首发在公众号，欢迎关注。

码农公寓

相关文章