FastAPI + Vue 实现 OAuth2 的 jwt token 登录验证

2021-09-13 21:01:49 阅读：385 来源： 互联网

标签：username access Vue jwt FastAPI redis db token user

后端使用 FastAPI，前端使用 Vue 来完成登录过程的用 jwt token 实现登录验证功能。

一、后端 FastAPI

1.1 设计用户表

本文采用的是 MySQL 数据库。
首先连接 MySQL 数据库，关于 MySQL 数据库的连接可参见另外一篇文章：FastAPI 连接 MySQL

用户的数据库表如下：

class User(Base):
    __tablename__ = 'users'  # 数据库表名

    username = Column(String(255), primary_key=True, nullable=False, unique=True, index=True)
    hashed_password = Column(String(255), nullable=False)
    name = Column(String(255))
    phone = Column(String(255), nullable=False)

在数据库表中添加一条记录：

username：test
hashed_password：$2b 12 12 12EixZaYVK1fsbw1ZfbX3OXePaWxn96p36WQoeG6Lruj3vjPGga31lW
name：test
phone：11111111111

对用户的查询功能实现：

from sqlalchemy.orm import Session

def get_user(db: Session, username: str) -> mysql_po.User:
    """
    根据username获取该用户
    """
    return db.query(mysql_po.User).filter(mysql_po.User.username == username).first()

1.2 连接 Redis

首先启动 redis：命令行中输入 redis-server。
在 python 中连接 redis：

async def get_redis() -> StrictRedis:
    """
    获取Redis对象
    """
    redis = StrictRedis(host=redis_conf.HOST,
                        port=redis_conf.PORT,
                        db=redis_conf.db,
                        password=redis_conf.PASSWORD)
    try:
        yield redis
    finally:
        redis.close()

1.3 实现验证类

采用 FastAPI 自带的 OAuth 实现即可：

from fastapi.security import OAuth2PasswordBearer
from pydantic import BaseSettings
from sqlalchemy.orm import Session
from passlib.context import CryptContext
from datetime import datetime, timedelta
from jose import jwt

class UserTokenConfig(BaseSettings):
    """对用户登录时处理token的相关配置"""
    SECRET_KEY: str = "09d25e094faa6ca2556c818166b7a9563b93f7099f6f0f4caa6cf63b88e8d3e7"  # 通过命令行 `openssl rand -hex 32` 可以生成该安全密钥
    ALGORITHM: str = "HS256"
    ACCESS_TOKEN_EXPIRE_MINUTES: int = 30

user_token_conf = UserTokenConfig()

__oauth2_scheme = OAuth2PasswordBearer(tokenUrl="api/login/token")
pwd_context = CryptContext(schemes=['bcrypt'], deprecated='auto')

def get_pwd_hash(pwd):
    return pwd_context.hash(pwd)

def authenticate_user(mysql_db: Session, username: str, password: str) -> Union[bool, mysql_po.User]:
    """
    验证用户合法性
    :return: 若验证成功则返回 User 对象；若验证失败则返回 False
    """
    user = get_user(mysql_db, username)
    if not user:
        return False
    if not pwd_context.verify(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, user_token_conf.SECRET_KEY, algorithm=user_token_conf.ALGORITHM)
    return encoded_jwt

1.4 controller 实现

from fastapi import APIRouter, Depends, HTTPException, status,
from redis import StrictRedis
from sqlalchemy.orm import Session

router = APIRouter()

class Token(BaseModel):
    code: int = 0
    access_token: str = Field(description='本次登录的token')
    token_type: str = Field(default='Bearer', description='token的类型，统一为 Bearer')

@router.post("/token",
                   response_model=Token,
                   summary='登录接口，获取 token',
                   description="""采用OAuth2.0认证协议，登录时获取用户的token，
                                使用 x-www-form-urlencoded 格式, 
                                在 Request Body 提交 username 和 password 即可获得本次用户登录的token,
                                并会被缓存到 Redis 中保持一定的时间段""")
async def login_for_access_token(form_data: OAuth2PasswordRequestForm = Depends(),
                                 mysql_db: Session = Depends(get_mysql_db),
                                 redis_db: StrictRedis = Depends(get_redis)):
    user = authenticate_user(mysql_db, form_data.username, form_data.password)  # 查询数据库进行用户验证
    if not user:
        raise HTTPException(
            status_code=status.HTTP_401_UNAUTHORIZED,
            detail="Incorrect username or password",
            headers={"WWW-Authenticate": "Bearer"},
        )
    # 根据 username 生成 token
    access_token_expires = timedelta(minutes=user_token_conf.ACCESS_TOKEN_EXPIRE_MINUTES)
    access_token = create_access_token(
        data={"sub": user.username}, expires_delta=access_token_expires
    )

    redis_db.set(access_token, user.username, ex=user_token_conf.ACCESS_TOKEN_EXPIRE_MINUTES * 60)  # 设置 redis_db 缓存

    return Token(
        code=0,
        access_token=access_token,
        token_type='Bearer'
    )

1.5 测试

运行 FastAPI 后进入 /docs 页面，会看到上面有一个 Authorize 用来测试认证过程：

输入用户名 test、密码 secret 后发现可以认证成功。

标签：username,access,Vue,jwt,FastAPI,redis,db,token,user
来源： https://blog.csdn.net/qq_45668004/article/details/120274979

本站声明： 1. iCode9 技术分享网（下文简称本站）提供的所有内容，仅供技术学习、探讨和分享；
2. 关于本站的所有留言、评论、转载及引用，纯属内容发起人的个人观点，与本站观点和立场无关；
3. 关于本站的所有言论和文字，纯属内容发起人的个人观点，与本站观点和立场无关；
4. 本站文章均是网友提供，不完全保证技术分享内容的完整性、准确性、时效性、风险性和版权归属；如您发现该文章侵犯了您的权益，可联系我们第一时间进行删除；
5. 本站为非盈利性的个人网站，所有内容不会用来进行牟利，也不会利用任何形式的广告来间接获益，纯粹是为了广大技术爱好者提供技术内容和技术思想的分享性交流网站。