FastAPI 学习之路（五十六）将token缓存到redis

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

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

# 见chapter18"""
-*- encoding=utf-8 -*-
Time: 2024/6/28 16:16
Author: lc
Email: 15101006331@163.com
File: chapter18.py
""""""
使用（哈希）密码和 JWT Bearer 令牌的 OAuth2
"""
from fastapi import FastAPI, Depends, status, HTTPException
from fastapi.security import OAuth2PasswordRequestForm, OAuth2PasswordBearerfrom pydantic import BaseModel
from typing import Optional
from jose import JWTError, jwt
from datetime import datetime, timedelta
from passlib.context import CryptContextSECRET_KEY = "09d25e094faa6ca2556c818166b7a9563b93f7099f6f0f4caa6cf63b88e8d3e7"
ALGORITHM = "HS256"
ACCESS_TOKEN_EXPIRE_MINUTES = 30oauth2_scheme = OAuth2PasswordBearer(tokenUrl="token")fake_db_users = {"mrli": {"username": "mrli","full_name": "mrli_hanjing","email": "mrli@qq.com","hashed_password": "$2b$12$EixZaYVK1fsbw1ZfbX3OXePaWxn96p36WQoeG6Lruj3vjPGga31lW","disabled": False}
}app = FastAPI()def fake_hash_password(password: str):"""模拟加密密码"""return passwordclass Token(BaseModel):access_token: strtoken_type: strclass TokenData(BaseModel):username: Optional[str] = Noneclass User(BaseModel):username: strfull_name: Optional[str] = Noneemail: Optional[str] = Nonedisabled: Optional[bool] = Noneclass UserInDB(User):hashed_password: strpwd_context = CryptContext(schemes=["bcrypt"], deprecated="auto")def verify_password(plain_password: str, hashed_password: str):"""校验密码"""return pwd_context.verify(plain_password, hashed_password)def get_password_hash(password: str):"""密码加密"""return pwd_context.hash(password)def get_user(db_users: dict, username: str):if username in db_users:user_info = db_users[username]return UserInDB(**user_info)def authenticate_user(db_users: dict, username: str, password: str):"""校验用户权限"""user = get_user(db_users, username)if not user:return Falseif not verify_password(password, user.hashed_password):return Falsereturn userdef create_access_token(data: dict, expires: Optional[timedelta] = None):"""创建jwt"""to_encode = data.copy()if expires:expire_time = datetime.utcnow() + expireselse:expire_time = datetime.utcnow() + timedelta(minutes=30)to_encode.update({"exp": expire_time})encode_jwt = jwt.encode(to_encode, SECRET_KEY, algorithm=ALGORITHM)return encode_jwtdef fake_decode_token(token):"""模拟解码token"""user = get_user(fake_db_users, token)return userdef get_current_user(token: str = Depends(oauth2_scheme)):exc = HTTPException(status_code=status.HTTP_401_UNAUTHORIZED,detail="Authentication Failed",headers={"WWW-Authenticate": "Bearer"})try:payload = jwt.decode(token, SECRET_KEY, algorithms=[ALGORITHM])username = payload.get("sub")if username is None:raise exctoken_data = TokenData(username=username)except JWTError:raise excuser = get_user(fake_db_users, username=token_data.username)if not user:raise excreturn userdef get_current_active_user(current_user: User = Depends(get_current_user)):if current_user.disabled:raise HTTPException(status_code=400, detail="Inactive user")return current_user@app.post("/token", response_model=Token)  # 必须实现路径为/token的接口来返回access_token，在文档页面点击Authorize时就是调用的这个接口
def login(form_data: OAuth2PasswordRequestForm = Depends()):user = authenticate_user(fake_db_users, 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"})access_token_expire = timedelta(minutes=ACCESS_TOKEN_EXPIRE_MINUTES)access_token = create_access_token(data={"sub": form_data.username}, expires=access_token_expire)return {"access_token": access_token,"token_type": "bearer"}@app.get("/me")
def get_me(current_user: User = Depends(get_current_active_user)):return current_user

我们需要把这部分代码进行整理，我们吊证到routers中的users.py。实际上就是把之前的方法柔和到新的方法中，需要调整下之前的创建用户，把登录实现了。

我们看下修改后的代码：

from fastapi import APIRouter, status
from fastapi import Depends, HTTPException
from starlette.requests import Requestfrom models.crud import *
from models.schemas import UserToken
from datetime import timedelta, datetime
from jose import JWTError, jwt
from typing import OptionalSECRET_KEY = "09d25e094faa6ca2556c818166b7a9563b93f7099f6f0f4caa6cf63b88e8d3e7"
ALGORITHM = "HS256"
ACCESS_TOKEN_EXPIRE_MINUTES = 30user_router = APIRouter()
from . import create_db
from passlib.context import CryptContext
pwd_context = CryptContext(schemes=["bcrypt"], deprecated="auto")def verify_password(plain_password: str, hashed_password: str):"""校验密码"""return pwd_context.verify(plain_password, hashed_password)def get_password_hash(password: str):"""密码加密"""return pwd_context.hash(password)def create_access_token(data: dict, expires: Optional[timedelta] = None):"""创建jwt"""to_encode = data.copy()if expires:expire_time = datetime.utcnow() + expireselse:expire_time = datetime.utcnow() + timedelta(minutes=30)to_encode.update({"exp": expire_time})encode_jwt = jwt.encode(to_encode, SECRET_KEY, algorithm=ALGORITHM)return encode_jwtdef get_current_user(token):exc = HTTPException(status_code=status.HTTP_401_UNAUTHORIZED,detail="Authentication Failed",headers={"WWW-Authenticate": "Bearer"})try:payload = jwt.decode(token, SECRET_KEY, algorithms=[ALGORITHM])username = payload.get("sub")if username is None:raise excreturn usernameexcept JWTError:raise exc@user_router.post("/user", response_model=UserOut)
def create_user(user: UserModel, db: Session = Depends(create_db)):return create_user_method(db, user)@user_router.get("/user", response_model=UserOut)
def get_user(uid: int, db: Session = Depends(create_db)):return get_user_method(db, uid)@user_router.post("/login")
async def login(request: Request, user: UserModel, db: Session = Depends(create_db)):db_user = get_user_by_email(db, user.email)pass

现在登录还未完全实现，接下来我们实现这个api。

这里的UserOut在schemas中实现。

class UserToken(BaseUser):token: str

登录的具体实现：

@user_router.post("/login")
async def login(request: Request, user: UserModel, db: Session = Depends(create_db)):db_user = get_user_by_email(db, user.email)# 密码校验verify = verify_password(user.password, db_user.hashed_password)if verify:# 产生tokentoken = create_access_token(data={"sub": user.email})is_cached = await request.app.state.redis.get(user.email)if is_cached:raise HTTPException(status_code=200, detail="请勿重复登录")await request.app.state.redis.set(user.email, token, expire=ACCESS_TOKEN_EXPIRE_MINUTES*60)user_token = UserToken(token=token, email=user.email)return user_tokenelse:raise HTTPException(status_code=200, detail="用户名或密码错误")

redis的相关操作还是使用上次分享时的startup和shutdown方法

我们启动测试，看是否正确，由于我们更改了创建用户时密码的hash方式，所以我们先创建用户

 接下来，我们调用登录api

这样token就产生了，redis中也有对应的缓存信息

通过本节，我们将登录的用户存储到了数据库中，并将登录后的用户token缓存到了redis，接下来我们将分享如何校验token 

本文来自互联网用户投稿，该文观点仅代表作者本人，不代表本站立场。本站仅提供信息存储空间服务，不拥有所有权，不承担相关法律责任。如若转载，请注明出处：http://xiahunao.cn/news/3247887.html

如若内容造成侵权/违法违规/事实不符，请联系瞎胡闹网进行投诉反馈，一经查实，立即删除！

相关文章

[Redis]典型应用——缓存

[Redis]典型应用——缓存

什么是缓存 缓存（Cache）是一种用于临时存储数据的机制，目的是提高数据访问速度和系统性能。 核心思路就是把一些常用的数据放到触手可及(访问速度更快)的地方，方便随时读取 缓存是一个相对的概念，比如说&#xff0c…
阅读更多...
域泛化（Domain Generalization）

域泛化（Domain Generalization）

仓库：https://github.com/jindongwang/transferlearning 综述：https://arxiv.org/pdf/2103.03097、https://arxiv.org/pdf/2103.02503 1.问题及解决方案 出发点：需要解决domain shift、out-of-distribution (OOD)问题 解决方案：绕…
阅读更多...
面试题整理 - 进程与线程问题

面试题整理 - 进程与线程问题

1.进程线程区别: 1.从本质上区分: 进程是操作系统资源分配的基本单位 线程是任务调度和执行的基本单位 2.在开销方面： 每个进程都有独立的代码和数据空间（程序上下文），程序之间的切换会有较大的开销 线程可以看做轻量级的进程&…
阅读更多...
爬虫案例（读书网）（下）

爬虫案例（读书网）（下）

上篇链接： CSDN-读书网https://mp.csdn.net/mp_blog/creation/editor/139306808 可以看见基本的全部信息：如(author、bookname、link.....) 写下代码如下： import requests from bs4 import BeautifulSoup from lxml import etreeheaders{…
阅读更多...
设计模式：真正的建造者模式

设计模式：真正的建造者模式

又臭又长的set方法 经常进行Java项目开发使用各类starter的你一定见过这种代码： public class SwaggerConfig {Beanpublic Docket api() {return new Docket(DocumentationType.SWAGGER_2).select().apis(RequestHandlerSelectors.any()).paths(PathSelectors.any…
阅读更多...
解决VMware虚拟机在桥接模式下无法上网的问题

解决VMware虚拟机在桥接模式下无法上网的问题

解决VMware虚拟机在桥接模式下无法上网的问题 windows11系统自动启动了热点功能，开启热点可能会干扰虚拟机的桥接设置。 方法一：windows11可以提供网络热点服务 方法二：手动指定桥接的物理网卡 方法一：关闭热点功能 优点&#xff…
阅读更多...
少儿编程启蒙宝典：Scratch动画游戏108变

少儿编程启蒙宝典：Scratch动画游戏108变

一、编程教育的时代价值与意义 随着数字时代的深入发展，社会对人才的需求正发生深刻变革，计算思维与编程能力已成为衡量个人竞争力的重要指标。在此背景下，培养孩子们运用计算思维解决实际问题的能力，成为教育领域的重要任务。编…
阅读更多...
运动用什么骨传导耳机好？推荐这五款运动骨传导耳机！

运动用什么骨传导耳机好？推荐这五款运动骨传导耳机！

在运动生涯，我见证了自我挑战与超越的每一个瞬间，而这一切都离不开那如影随形的运动骨传导耳机。一款出色的运动耳机，其重要性不言而喻——它不仅是提升运动效率的得力助手，更是开启多元化运动体验的金钥匙。近年来，运…
阅读更多...
网络结构-组件-AI（九）

网络结构-组件-AI（九）

深度学习网络组件 RNN公式讲解计算示意图讲解 CNN计算示意 Normalization(归一化层)Normalization常见两种方式 Dropout层 RNN 循环神经网络（recurrent neural network） 主要思想： 即将整个序列划分成多个时间步，将每一个时间步的…
阅读更多...
创建通用JS公共模块并发布至npm

创建通用JS公共模块并发布至npm

title: 创建通用JS公共模块并发布至npm tags: UMD rollup verdaccio npm categories: 模块化 概要内容 创建：JS公共模块 打包：使用rollup 打包公共模块 发布：js公共模块至verdaccio平台 发布：js公共模块至npm平台 如何创建JS公共模…
阅读更多...
媒体邀约宣传做了13年，我们总结了哪些经验？

媒体邀约宣传做了13年，我们总结了哪些经验？

传媒如春雨，润物细无声，大家好，我是51媒体网胡老师。 「51媒体」作为一家在媒体邀约宣传领域深耕13年的专业机构，积累了一些经验。现在与大家分享下： 合理的制定媒体邀约传播方案 在进行媒体邀约前，首先需…
阅读更多...
木舟0基础学习Java的第二十天(线程，实现，匿名有名，休眠，守护，加入，设计，计时器，通信)

木舟0基础学习Java的第二十天(线程，实现，匿名有名，休眠，守护，加入，设计，计时器，通信)

多线程 并发执行的技术 并发和并行 并发：同一时间 有多个指令 在单个CPU上 交替执行 并行：同一时间 有多个指令 在多个CPU上 执行 进程和线程 进程：独立运行 任何进程 都可以同其他进程一起 并发执行 线程：是进程中的单个顺…
阅读更多...
【人工智能】深度剖析AI伦理：强化隐私防线，推动算法公平性的核心议题

【人工智能】深度剖析AI伦理：强化隐私防线，推动算法公平性的核心议题

文章目录 🍊1 人工智能兴起背后的伦理及道德风险1.1 算法偏见与歧视1.2 数据隐私侵权1.3 透明度受限1.4 决策失衡1.5 AI生成内容的危险性 🍊2 建构AIGC伦理观：实现人机共创的永续提升2.1 技术手段与伦理预防2.2 即时警告与紧急关停措施2.3 法…
阅读更多...
图片如何去水印，PS 图片去水印的几种常见方法

图片如何去水印，PS 图片去水印的几种常见方法

在数字图像的世界里，水印常常被用来标识版权或防止未经授权的使用，但有时它们却成为了美观的障碍。无论是出于个人偏好还是专业需求，去除图片上的水印已经成为一项常见的任务。 Adobe Photoshop 作为行业标准的图像编辑软件，提供…
阅读更多...
队列（Queue），循环队列，双端队列（Deque）and LeetCode刷题

队列（Queue），循环队列，双端队列（Deque）and LeetCode刷题

队列（Queue），循环队列，双端队列（Deque）and LeetCode刷题 1. 队列的概念2.队列的使用3. 队列的模拟实现3.1 用链式结构实现队列3.2 用顺序结构实现队列 4. 循环队列5. 双端队列（Deque&#xff09…
阅读更多...
【内网安全】横向移动-Wmi-Smb-CME密码喷射

【内网安全】横向移动-Wmi-Smb-CME密码喷射

目录 环境介绍域信息收集-横向移动前置判断是不是在域内获取域控主机的内网ip端口扫描内网获取主机密码 域横向移动-WMI-自带&命令&套件&插件1.wmic系统自带：(单执行：即无回显) 2.cscript系统自带：(交互式) 3.wmiexec-impacket&a…
阅读更多...
文献阅读：A Case for Managed and Model-less Inference Serving

文献阅读：A Case for Managed and Model-less Inference Serving

目录 知识点记录推理服务在线推理特点 动机：为什么作者想要解决这个问题？贡献：作者在这篇论文中完成了什么工作(创新点)？规划：他们如何完成工作？1.挑战1.1 选择一个模型变体1.2 异构硬件1.3 资源提供1.4 启…
阅读更多...
MySQL双主双从实现方式

MySQL双主双从实现方式

双主双从（MM-SS） 前言 避免单一主服务器宕机，集群写入能力缺失 从 1 复制 主1 ，从 2 复制 主 2 主 1 复制 主 2，主 2 复制主 1 也就是 主 1 和主 2 互为主从。主1主2互为主从， 是为了以下情景&#xff0c…
阅读更多...
初识XXE漏洞及ctfshow做题（373-378）

初识XXE漏洞及ctfshow做题（373-378）

初识XXE漏洞 1.XXE简介 XXE就是XML外部实体注入，当允许引用外部实体时， XML数据在传输中有可能会被不法分子被修改，如果服务器执行被恶意插入的代码，就可以实现攻击的目的攻击者可以通过构造恶意内容，就可能导致任意…
阅读更多...
昇思25天学习打卡营第29天 | 文本解码原理--以MindNLP为例

昇思25天学习打卡营第29天 | 文本解码原理--以MindNLP为例

今天是29天，学习了文本解码原理--以MindNLP为例。 MindNLP 是一个基于 MindSpore 的开源自然语言处理（NLP）库。它具有以下特点： 支持多种 NLP 任务：如语言模型、机器翻译、问答、情感分析、序列标记、摘要等&#xff…
阅读更多...
最新文章

Copyright @ 2022~2023