为flask项目配置redis来缓存验证码
·
1.下载redis的服务器端,现在本地运行,因为没云服务器
windows版本,官网没有win版
参考文章Window下Redis的安装和部署详细图文教程(Redis的安装和可视化工具的使用)_redis安装-CSDN博客2. 在项目环境下下载redis客户端,pip install redis
3.先建一个文件夹utils用来保存工具,记得建立一个__init__.py文件,导包的时候python会将其加入搜索路径,在目录下建立redis_config.py文件,用来设置redis的配置信息,建立了redis客户端redis_client
#配置redis服务器的基本参数,用于验证码存储和定期销毁
#pip install redis安装redis客户端
import redis
# 创建Redis连接池,用于持续连接,每次连接不需要断开
redis_pool = redis.ConnectionPool(
host='localhost', # Redis服务器地址
port=6379, # Redis服务器端口
db=0, # 使用的数据库编号
decode_responses=True # 返回字符串而非字节
)
# 创建Redis连接对象
redis_client = redis.Redis(connection_pool=redis_pool)
4.之后在utils目录下建立captcha.py文件,里面书写保存和验证验证码的操作函数
# Redis 验证码工具类
from utils.redis_config import redis_client
import time
# 验证码默认有效期(5分钟)
CAPTCHA_EXPIRE_SECONDS = 5 * 60
def save_captcha(email: str, captcha: str):
"""保存验证码到Redis"""
# 使用邮箱作为key,验证码作为value
redis_client.setex(email, CAPTCHA_EXPIRE_SECONDS, captcha)
def verify_captcha(email: str, user_input: str) -> bool:
"""验证验证码是否有效"""
# 获取存储的验证码
stored_captcha = redis_client.get(email)
# 检查验证码是否存在
if not stored_captcha:
return False
# 验证验证码是否匹配
is_valid = stored_captcha == user_input
# 如果验证通过,删除验证码(一次性使用)
if is_valid:
redis_client.delete(email)
return is_valid
def delete_captcha(email: str):
"""删除指定邮箱的验证码"""
redis_client.delete(email)
def clear_expired_captchas():
"""清理所有过期的验证码(Redis会自动处理)"""
# Redis会自动删除过期的key,不需要手动清理
pass
5.在auth.py文件中更改用户点击获取验证码之后的操作逻辑,用户点击后会在后端产生验证码,并用redis暂存验证码,具体逻辑在captcha/email中
#登录功能
#从flask中导入蓝图对象
import random
from flask import Blueprint
#渲染登录页面模板
from flask import render_template
#导入格式化邮件信息的包
from flask_mail import Message
#导入邮件发送器
from exts import mail,db
#查询字符串获取参数
from flask import request
#原始可供提取验证码的字符串
import string
#从原始字符串中提取出任意几位的验证码
import random
#将字典或者列表转换为json响应
from flask import jsonify
#导入数据库对象
from models import EmailCaptchaModel
#从表单验证模块forms.py中导入表单验证功能
from .forms import RegisterForm
#从models导入数据库写入库
from models import UserModel
#导入页面跳转包
from flask import redirect
#导入redis中存储验证码的函数
from utils.captcha_redis import save_captcha
bp = Blueprint("auth",__name__,url_prefix="/auth")
#/auth/login
@bp.route("/login")
def login():
return "what"
#GET获取页面
#POST提交注册数据默认是GET
@bp.route("/register",methods=['GET','POST'])
def register():
if request.method == 'GET':
return render_template("register.html")
else:
#提交注册数据后,1.验证 2.验证通过转向登录界面,不通过返回注册界面
#表单验证 flask_wtf wtforms
#从forms.py中导入表单验证的部分验证数据是否正确
form = RegisterForm(request.form)
#如果验证通过就存储
if form.validate():
email = form.email.data
username = form.username.data
password = form.password.data
#将注册用户数据存储进数据库
user = UserModel(email=email,username=username,password=password)
db.session.add(user)
db.session.commit()
#转向登录界面,通过视图函数找到url跳转
return redirect("/auth/login")#第一种跳转方法,路径跳转法
# return redirect(url_for("auth.login"))#蓝图跳转法
else:#表单验证失败,重点位到注册页面
print(form.errors)
return "失败"
#当用户点击获取验证码的button时候,会在register.js中触发这个事件,然后执行下面的函数,发送验证码,并存储在数据库中
@bp.route("/captcha/email")
def get_email_captcha():
#变量传参: /captcha/email/<email>
#查询字符串:/captcha/email?email=xxx@qq.com
email = request.args.get("email")
if not email:
return jsonify({"code": 400, "message": "邮箱不能为空"}), 400
#验证码格式是4到6位的数字字母组合
source = string.digits*4#0123456789012345678901234567890123456789
captcha = random.sample(source,4)#从source中随机提取出4位字符,组成一个list
captcha = "".join(captcha)#以“”为分隔符合并列表
message = Message(subject="知乎注册验证码",recipients=[email],body=f"您的验证码是{captcha}")#写邮件,注意获取查询字符串的方式使用中括号
mail.send(message)#发送邮件
#使用数据库存储验证码用于验证,有memcached/redis两种暂存方法,这里使用数据库存储
# email_captcha = EmailCaptchaModel(email=email,captcha=captcha)#赋值,一条记录,id不用写出来
# db.session.add(email_captcha)
# db.session.commit() #在terminal输入三个步骤进行数据库更新,1.flask db init (只执行一次) 2.flask db migrate 3.flask db upgrade
#使用redis存储验证码1.from captcha_redis import save_captcha 2. save_captcha(email, captcha)
save_captcha(email,captcha)#redis使用内存存储
return jsonify({"code": 200,"message":"","data": None})
#测试用途,没有参与项目
@bp.route("/mail")
def mail_test():
message = Message(subject="邮箱测试",recipients=["3125988872@qq.com"],body="这是一条测试信息")
mail.send(message)
return "邮件发送成功"
6.在forms.py文件中更改验证码的验证逻辑,将用户提交的验证码和redis中暂存的验证码进行比对
#表单验证
#包:pip install flask_wtf,会自动安装wtforms
import wtforms
#从wtforms.validators中导入检查邮件格式,验证码格式,的包
from wtforms.validators import Email,Length,EqualTo
#导入数据表模型,查看邮箱是否已经注册
from models import UserModel
#导入数据表,查询临时存储的邮箱和验证码的映射是否正确
from models import EmailCaptchaModel
#导入数据库
from exts import db
#导入时间库
from datetime import datetime,timedelta
#导入redis的处理函数from captcha_redis import verify_captcha, delete_captcha
from utils.captcha_redis import verify_captcha, delete_captcha
#Form: 用于验证前端提交的注册数据是否正确
#1.验证邮箱格式 ,验证码,密码,确认密码格式 2.邮箱是否已经注册 3。邮箱和验证码必须双正确
class RegisterForm(wtforms.Form):
#
email = wtforms.StringField(validators=[Email(message="邮件格式错误")])
captcha = wtforms.StringField(validators=[Length(min=4,max=4,message="验证码格式错误!")])
username = wtforms.StringField(validators=[Length(min=3,max=20,message="用户名格式错误")])
password = wtforms.StringField(validators=[Length(min=6,max=20,message="密码格式错误")])
password_confirm = wtforms.StringField(validators=[EqualTo("password")])
#自定义验证
#1.邮箱是否已经注册
def validate_email(self, field):
email = field.data#field指代上面的email对象,field.data获得属性
# 注意:这里应该查询UserModel而不是EmailCaptchaModel
user = UserModel.query.filter_by(email=email).first()#根据email这个值去查询表,看是否有记录相等
if user:#存在,重复,不能重复注册
raise wtforms.ValidationError(message="该邮箱已经被注册")
#不存在就不做处理
#2.验证验证码是否正确,邮箱必须和验证码一一对应
def validate_captcha(self,field):
captcha = field.data
email = self.email.data
# captcha_model = EmailCaptchaModel.query.filter_by(email=email,captcha=captcha),不加first是查询对象,不能被修改
# 执行查询获取实际的模型实例
# captcha_model = EmailCaptchaModel.query.filter_by(email=email, captcha=captcha).first()
# if not captcha_model:#如果查不到记录,则错误
# raise wtforms.ValidationError(message="邮箱或者验证码错误")
# else:#正确,删除验证码记录
# # todo : 可以删除captcha_model
# db.session.delete(captcha_model)
# db.session.commit()
#查到验证码,验证码不能过期
# now = datetime.now()
# expire_time = captcha_model.create_time + timedelta(minutes=5)#总经历的时间
#
# if now > expire_time:
#验证码过期,同时删除当前验证码记录
# EmailCaptchaModel.query.filter_by(email=email,captcha=captcha).delete()
# db.session.commit()
# raise wtforms.ValidationError(message="验证码已过期,请重新获取")
#验证码有效,删除该邮箱对应的所有验证码记录
# try:
# EmailCaptchaModel.query.filter_by(email=email).delete()
# db.session.commit()
# except Exception as e:
# db.session.rollback()
# raise wtforms.ValidationError(message="验证码处理失败,请重试")
#使用redis验证验证码
#1.导入redis的处理函数from captcha_redis import verify_captcha, delete_captcha
#2.验证验证码是否正确且未过期
# 使用Redis验证验证码
if not verify_captcha(email, captcha):
raise wtforms.ValidationError(message="邮箱或者验证码错误")
# 验证通过后删除验证码
delete_captcha(email)
总结:redis服务器->redis客户端(pip install redis)->redis服务器配置(redis.config)->建立redis客户端(redis_client)->前端请求验证码->后端产生验证码并暂存在redis中->前端收到验证码填写发到后端->后端在redis中比对结果,正确删除,错误报错,redis定期删除,本例是5*60,单位是秒
更多推荐




所有评论(0)