FastAPI 学习之路（五十六）将token存放在redis

2023-08-29 23:02:52

在之前的文章中，FastAPI 学习之路（二十九）使用（哈希）密码和 JWT Bearer 令牌的 OAuth2，FastAPI 学习之路（二十八）使用密码和 Bearer 的简单 OAuth2，FastAPI 学习之路（三十四）数据库多表操作，我们分享了基于jwt认证token和基于数据库创建用户，那么我们今天把这些代码整理下，形成基于数据库用户名密码，登陆验证token存储到redis中。

  首先我们看下之前基于jwt认证token的代码

from fastapi import  Depends,status,HTTPException
from pydantic import BaseModel
from typing import Optional
from jose import JWTError, jwt
from datetime import datetime, timedelta
from passlib.context import CryptContext
SECRET_KEY = "09d25e094faa6ca2556c818166b7a9563b93f7099f6f0f4caa6cf63b88e8d3e7"
ALGORITHM = "HS256"
ACCESS_TOKEN_EXPIRE_MINUTES = 30
# oauth2_scheme = OAuth2PasswordBearer(tokenUrl="token")
fake_users = {
    "leizi": {
        "username": "leizi",
        "full_name": "leizishuoceshikaifa",
        "email": "leizi@leizi.com",
        "hashed_password": "$2b$12$4grMcfV9UMijFC0CEeJOTuTHE21msQOmkUWuowUewRSXt8cimW/76",
        "disabled": False
    }
}
def fake_hash_password(password: str):
    return password

class Token(BaseModel):
    access_token: str
    token_type: str
class TokenData(BaseModel):
    username: Optional[str] = None
    password:Optional[str]=None
class User(BaseModel):
    username: str
    email: Optional[str] = None
    full_name: Optional[str] = None
    disabled: Optional[bool] = None
class UserInDB(User):
    hashed_password: str
pwd_context = CryptContext(schemes=["bcrypt"], deprecated="auto")
def verify_password(plain_password, hashed_password):
    return pwd_context.verify(plain_password, hashed_password)
def get_password_hash(password):
    return pwd_context.hash(password)
def authenticate_user(fake_db, username: str, password: str):
    user = get_user(fake_db, username)
    print(get_password_hash(password))
    if not user:
        return False

    if not verify_password(password, user.hashed_password):
        return False
    return user
def create_access_token(data: dict, expires_delta: Optional[timedelta] = None):
    to_encode = data.copy()
    if expires_delta:
        expire = datetime.utcnow() + expires_delta
    else:
        expire = datetime.utcnow() + timedelta(minutes=15)
    to_encode.update({"exp": expire})
    encoded_jwt = jwt.encode(to_encode, SECRET_KEY, algorithm=ALGORITHM)
    return encoded_jwt
def get_user(db, username: str):
    if username in db:
        user_dict = db[username]
        return UserInDB(**user_dict)
def fake_decode_token(token):
    user = get_user(fake_users, token)
    return user


def get_current_user(token: str = Depends()):
    credentials_exception = HTTPException(
        status_code=status.HTTP_401_UNAUTHORIZED,
        detail="验证失败",
        headers={"WWW-Authenticate": "Bearer"},
    )
    try:
        payload = jwt.decode(token, SECRET_KEY, algorithms=[ALGORITHM])
        username: str = payload.get("sub")
        if username is None:
            raise credentials_exception
        token_data = TokenData(username=username)
    except JWTError:
        raise credentials_exception
    user = get_user(fake_users, username=token_data.username)
    if user is None:
        raise credentials_exception
    return user
def get_current_active_user(current_user: User = Depends(get_current_user)):
    if current_user.disabled:
        raise HTTPException(status_code=400, detail="已经删除")
    return current_user
@app.post("/login", response_model=Token)
async def login_for_access_token( tokendata:TokenData):
    user = authenticate_user(fake_users,tokendata.username,tokendata.password)
    if not user:
        raise HTTPException(
            status_code=status.HTTP_401_UNAUTHORIZED,
            detail="Incorrect username or password",
            headers={"WWW-Authenticate": "Bearer"},
        )
    access_token_expires = timedelta(minutes=ACCESS_TOKEN_EXPIRE_MINUTES)
    access_token = create_access_token(
        data={"sub": user.username}, expires_delta=access_token_expires
    )
    return {"access_token": access_token, "token_type": "bearer"}

  我们需要把这部分代码进行调整，我们调整到routers中的user.py。其实就是把之前的方法去柔和到新的方法中，需要调整下之前的用户创建，把登陆给实现了。

    我们看下新修改后的代码。

from fastapi import APIRouter,status
from fastapi import  Depends,HTTPException
from models.crud import *
from get_db import get_db
from  datetime import timedelta,datetime
from jose import JWTError, jwt
usersRouter=APIRouter()
SECRET_KEY = "09d25e094faa6ca2556c818166b7a9563b93f7099f6f0f4caa6cf63b88e8d3e7"
ALGORITHM = "HS256"
ACCESS_TOKEN_EXPIRE_MINUTES = 30


from passlib.context import CryptContext
pwd_context = CryptContext(schemes=["bcrypt"], deprecated="auto")
def verify_password(plain_password, hashed_password):
    return pwd_context.verify(plain_password, hashed_password)
def get_password_hash(password):
    return pwd_context.hash(password)
# 新建用户
@usersRouter.post("/users/", tags=["users"], response_model=Users)
def create_user(user: UserCreate, db: Session = Depends(get_db)):
    """
        - **email**: 用户的邮箱
        - **password**: 用户密码
        """
    db_crest = get_user_emai(db, user.email)
    user.password=get_password_hash(user.password)
    if not db_crest:
        return db_create_user(db=db, user=user)
    raise HTTPException(status_code=200, detail="账号不能重复")
def create_access_token(data: dict):
    to_encode = data.copy()
    encoded_jwt = jwt.encode(to_encode, SECRET_KEY, algorithm=ALGORITHM)
    return encoded_jwt
def get_cure(token):
    credentials_exception = HTTPException(
        status_code=status.HTTP_401_UNAUTHORIZED,
        detail="验证失败",
        headers={"WWW-Authenticate": "Bearer"},
    )
    try:
        payload = jwt.decode(token, SECRET_KEY, algorithms=[ALGORITHM])
        username: str = payload.get("sub")
        if username is None:
            raise credentials_exception
        return  username
    except JWTError:
        raise credentials_exception

@usersRouter.post("/login",response_model=UsersToken)
def login(user:UserCreate,db: Session = Depends(get_db)):
    db_crest = get_user_emai(db, user.email)
    if not db_crest:
        raise HTTPException(status_code=200, detail="账号不存在")
    pass

 现在登陆还未完全实现，我们去实现下这块。

        这里的UsersToken在schemas中实现。

class UsersToken(UserBase):
    token: str

    登陆的实现我们实现如下

@usersRouter.post("/login",response_model=UsersToken)
async def login(request: Request,user:UserCreate,db: Session = Depends(get_db)):
    #查看用户是否存在
    db_crest = get_user_emai(db, user.email)
    if not db_crest:
        raise HTTPException(status_code=200, detail="账号不存在")
    #校验密码
    verifypassowrd=verify_password(user.password,db_crest.password)
    if verifypassowrd:
        #产生token
        token=create_access_token(data={"sub": user.email})
        useris=await request.app.state.redis.get(user.email)
        if not useris:
            request.app.state.redis.set(user.email,token,expire=ACCESS_TOKEN_EXPIRE_MINUTES*60)
            usertoken=UsersToken(token=token,email=user.email)
            return usertoken
        raise HTTPException(status_code=200, detail="请勿重复登陆")
    else:
        raise HTTPException(status_code=200, detail="密码错误")

 redis相关的还是在我们上次分享的时候的FastAPI 学习之路（五十四）操作Redis。

        我们去启动下去测试下，看我们实现的是否正确。

        由于我们更新了我们的创建用户的时候的密码的hash呢，我们先去创建用户

FastAPI 学习之路（五十六）将token存放在redis

 

  接下来，我们调用我们的登录

FastAPI 学习之路（五十六）将token存放在redis

 

 

发现登陆报错了。

FastAPI 学习之路（五十六）将token存放在redis

 

 

        这里我们在设计数据库的时候用的是hashed_password存储的密码，我们这里需要修改下

    verifypassowrd=verify_password(user.password,db_crest.hashed_password)

  然后我们在测试下

FastAPI 学习之路（五十六）将token存放在redis

 

  这样我们的token就产生了，我们也在redis有了存储

FastAPI 学习之路（五十六）将token存放在redis

 

 

那么接下来会分享如何校验token？

        通过本次的分享，我们讲登陆的用户存储到了数据库中，讲登陆后的产生的token我们存储到了redis上了。这样我们的存储持久化，接下来，我会分享如何校验token做判断是否登陆。

        所有的代码，都会放在gitee上，大家可以后续看到完整的代码。后续将开发几个接口，和结合我们的接口测试来分享。欢迎持续关注。

https://gitee.com/liwanlei/fastapistuday

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

FastAPI 学习之路（五十六）将token存放在redis

上一篇：【FastAPI 学习 二】SqlAlchemy Model模型类


下一篇：(完结篇)Python框架FastAPI：比Flask和Tornada更高性能的API 框架