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://www.mzph.cn/news/873928.shtml

如若内容造成侵权/违法违规/事实不符，请联系多彩编程网进行投诉反馈email:809451989@qq.com，一经查实，立即删除！

相关文章

Kubernetes 之 Ingress

Kubernetes 之 Ingress

Kubernetes 之 Ingress 定义 Ingress 可以把外部需要进入到集群内部的请求转发到集群中的一些服务上，从而实现把服务映射到集群外部的需要。Ingress 能把集群内 Service 配置成外网能够访问的 URL，流量负载均衡，提供基于域名访问的虚拟主机…
阅读更多...
RabbitMQ 和 RocketMQ 的区别

RabbitMQ 和 RocketMQ 的区别

RabbitMQ 和 RocketMQ 都是流行的开源消息中间件，它们用于在分布式系统中异步传输消息。尽管它们都实现了核心的消息队列功能，但它们在设计、性能、特性和使用场景上有一些关键的区别： 基础架构: RabbitMQ: 基于AMQP（高级消息队列…
阅读更多...
阵列信号处理学习笔记（二）--空域滤波基本原理

阵列信号处理学习笔记（二）--空域滤波基本原理

阵列信号 阵列信号处理学习笔记（一）–阵列信号处理定义 阵列信号处理学习笔记（二）–空域滤波基本原理 文章目录 阵列信号前言一、阵列信号模型1.1 信号的基本模型1.2 阵列的几何构型1.3 均匀直线阵的阵列信号基本模型 总结 前言…
阅读更多...
HOW - React 处理不紧急的更新和渲染

HOW - React 处理不紧急的更新和渲染

目录 useDeferredValueuseTransitionuseIdleCallback 在 React 中，有一些钩子函数可以帮助你处理不紧急的更新或渲染，从而优化性能和用户体验。 以下是一些常用的相关钩子及其应用场景： useDeferredValue 用途：用于处理高优先级…
阅读更多...
嵌入式面试总结

嵌入式面试总结

C语言中struct和union的区别 struct和union都是常见的复合结构。 结构体和联合体虽然都是由多个不同的数据类型成员组成的，但不同之处在于联合体中所有成员共用一块地址空间，即联合体只存放了一个被选中的成员，结构体中所有成员占用空间是累…
阅读更多...
【网络】windows和linux互通收发

【网络】windows和linux互通收发

windows和linux互通收发 一、windows的udp客户端代码1、代码剖析2、总体代码 二、linux服务器代码三、成果展示 一、windows的udp客户端代码 1、代码剖析 首先我们需要包含头文件以及lib的一个库&#xff1a; #include <iostream> #include <WinSock2.h> #inclu…
阅读更多...
前端页面是如何禁止被查看源码、被下载，被爬取，以及破解方法

前端页面是如何禁止被查看源码、被下载，被爬取，以及破解方法

文章目录 1.了解禁止查看，爬取原理1.1.JS代码，屏蔽屏蔽键盘和鼠标右键1.2.查看源码时，通过JS控制浏览器窗口变化2.百度文库是如何防止抓包2.1.HTPPS2.2. 动态加载为什么看不到？如何查看动态加载的内容？3.禁止复制，如果解决3.1.禁止复制原理3.2.如何破解1.了解禁止查看，爬…
阅读更多...
使用scikit-learn进行机器学习：基础教程

使用scikit-learn进行机器学习：基础教程

使用scikit-learn进行机器学习&#xff1a;基础教程 Scikit-learn是Python中最流行的机器学习库之一。它提供了简单易用的工具&#xff0c;帮助我们进行数据预处理、模型训练、评估和预测。本文将带你通过一个基础教程&#xff0c;了解如何使用scikit-learn进行机器学习。 1.…
阅读更多...
【模板代码】用于编写Threejs Demo的模板代码

【模板代码】用于编写Threejs Demo的模板代码

基础模板代码 使用须知常规模板代码常规Shader模板代码 使用须知 本模板代码&#xff0c;主要用于编写Threejs的Demo&#xff0c;因为本人在早期学习的过程中&#xff0c;大量抄写Threejs/examples下的代码以及各个demo站的代码&#xff0c;所以养成了编写Threejs的demo的习惯…
阅读更多...
SAP 采购订单 Adobe 消息输出

SAP 采购订单 Adobe 消息输出

目录 1 简介 2 业务数据例子 3 选择增强 & 代码 1&#xff09;BADI: MM_PUR_S4_PO_MODIFY_HEADER 2&#xff09;BADI: MM_PUR_S4_PO_MODIFY_ITEM 4 自定义 Adobe form 1&#xff09;PO Master form 2&#xff09;PO form 5 前台主数据配置 6 后台配置 1&#xf…
阅读更多...
昇思22天

昇思22天

CycleGAN图像风格迁移互换 CycleGAN&#xff08;循环生成对抗网络&#xff09;是一种用于在没有成对训练数据的情况下学习将图像从源域 X 转换到目标域 Y 的方法。该技术的一个重要应用是域迁移&#xff0c;即图像风格迁移。 模型介绍 模型简介: CycleGAN 来自于论文 Unpair…
阅读更多...
掌握Rust：函数、闭包与迭代器的综合运用

掌握Rust：函数、闭包与迭代器的综合运用

掌握Rust&#xff1a;函数、闭包与迭代器的综合运用 引言&#xff1a;解锁 Rust 高效编程的钥匙函数定义与模式匹配&#xff1a;构建逻辑的基石高阶函数与闭包&#xff1a;代码复用的艺术迭代器与 for 循环&#xff1a;高效数据处理的引擎综合应用案例&#xff1a;构建一个简易…
阅读更多...
Mybatis——一对多处理

Mybatis——一对多处理

环境搭建 与多对一相似&#xff0c;有一些地方需要改动&#xff1a; 实体类&#xff1a; Data public class Student {private int id;private String name;private int tid;} Data public class Teacher {private int id;private String name;// 一个老师拥有多个学生priv…
阅读更多...
【LeetCode】day15：110 - 平衡二叉树, 257 - 二叉树的所有路径, 404 - 左叶子之和, 222 - 完全二叉树的节点个数

【LeetCode】day15：110 - 平衡二叉树, 257 - 二叉树的所有路径, 404 - 左叶子之和, 222 - 完全二叉树的节点个数

LeetCode 代码随想录跟练 Day15 110.平衡二叉树257.二叉树的所有路径404.左叶子之和222.完全二叉树的节点个数 110.平衡二叉树 题目描述&#xff1a; 给定一个二叉树&#xff0c;判断它是否是 平衡二叉树 平衡二叉树的定义是&#xff0c;对于树中的每个节点&#xff0c;其左右…
阅读更多...
qt自定义控件（QLabel）

qt自定义控件（QLabel）

先创建自定义控件类painter_label 1.自定义类必须给基类传入父窗口指针 2.重写控件中的方法 3.在UI中创建一个QLabel,右键“提升为”&#xff0c;输入类名
阅读更多...
动画革命：Lottie如何改变我们对移动应用交互的认知

动画革命：Lottie如何改变我们对移动应用交互的认知

在数字世界的浩瀚星空中&#xff0c;每一个像素都跃动着无限创意与想象的火花。当静态的界面遇上动态的魔法&#xff0c;一场视觉盛宴便悄然开启。今天&#xff0c;让我们一同揭开一位幕后英雄的神秘面纱——Lottie&#xff0c;这个在UI/UX设计界掀起波澜的动画利器&#xff0c…
阅读更多...
掌握构建自动化：如何在Gradle中使用Init脚本进行构建初始化

掌握构建自动化：如何在Gradle中使用Init脚本进行构建初始化

掌握构建自动化&#xff1a;如何在Gradle中使用Init脚本进行构建初始化 在现代软件开发中&#xff0c;自动化构建是提高效率和一致性的关键。Gradle&#xff0c;作为一个功能强大的构建工具&#xff0c;提供了丰富的自动化支持。其中&#xff0c;Init脚本是Gradle中用于初始化…
阅读更多...
SVN与Git功能差异对比分析

SVN与Git功能差异对比分析

最近在调研学习Git管理和分支模型相关内容&#xff0c;外延到了SVN和Git差异、工作原理等相关细节&#xff0c;学习整理如下。 SVN&#xff08;Subversion&#xff09;与 Git 的最大不同&#xff0c;主要包括以下几个方面&#xff1a; 交流探讨&#xff0c;加入群聊【Java学习…
阅读更多...
51.2T 800G 以太网交换机，赋能AI开放生态

51.2T 800G 以太网交换机，赋能AI开放生态

IB与以太之争 以太网替代IB趋势明显。据相关报告&#xff1a;2024年TOP500的超算中&#xff0c;采用以太网方案占比48.5%&#xff0c;InfiniBand占比为39.2%&#xff0c;其中排名前6的超算中已有5个使用以太网互联。 开放系统战胜封闭系统仅是时间问题。我们已经看到&#xf…
阅读更多...
钡铼EdgeIO系统BL206对接MQTT、Modbus TCP、OPC UA

钡铼EdgeIO系统BL206对接MQTT、Modbus TCP、OPC UA

钡铼EdgeIO系统BL206提供双网口支持交换机级联功能&#xff0c;支持标准MQTT协议、Modbus TCP协议、OPC UA协议&#xff0c;由耦合器与IO模块组成&#xff0c;采用Web配置&#xff0c;内置云驱动、可编程逻辑控制功能&#xff0c;用户点击即可连接云平台。耦合器自带诊断功能&a…
阅读更多...
最新文章

Copyright @ 2022~2023