在之前的文章中，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呢，我们先去创建用户
接下来，我们调用我们的登录
发现登陆报错了。
这里我们在设计数据库的时候用的是hashed_password存储的密码，我们这里需要修改下
verifypassowrd=verify_password(user.password,db_crest.hashed_password)
然后我们在测试下
这样我们的token就产生了，我们也在redis有了存储
那么接下来会分享如何校验token？
通过本次的分享，我们讲登陆的用户存储到了数据库中，讲登陆后的产生的token我们存储到了redis上了。这样我们的存储持久化，接下来，我会分享如何校验token做判断是否登陆。
所有的代码，都会放在gitee上，大家可以后续看到完整的代码。后续将开发几个接口，和结合我们的接口测试来分享。欢迎持续关注。
https://gitee.com/liwanlei/fastapistuday
