在之前的文章中，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"}