---恢复内容开始---
一.python 使用redis
1.1 安装 pip install redis
测试有一些基本的数据类型
import redis # redis 是一个缓存数据库 # r = redis.Redis(host='127.0.0.1', port=6379) # 一般采用host与redis 可以不用配置， 用db来选择操作的数据库 # 如： r = redis.Redis(db=11) # r.set() 设置值 操作字符串 r.set('name','coco') print(r.get('name')) # 设置过期时间 # r.setex('age', 5, '18') # 不能一置打开代开 会重新设置 print(r.get('age')) # 5 秒后None
hset() 哈希字典
# hset # print(r.get('age')) # 5 秒后None r.hset('student','name','yye') # name key value 哈希字典的设置值和取值方法 print(r.hget('student','name'))
zdd() >>>zset（） sorted set() 有序集合
# zadd() # r.zadd('mayup',{'koko':100,'coco':60,'yye':80}) # 有序集合 # print(r.zrange('mayup',0,0)) # 正向取值 [b'coco'] # print(r.zrevrange('mayup',0,0)) # 取得最后一个 [b'koko']
1.2 redis 建立连接池
(1)第一种方式
# redis 原生的建立连接池(支持并发) pool = redis.ConnectionPool(db=12,max_connections=100) rp = redis.Redis(connection_pool=pool) rp.set('name','koko') print(rp.get('name'))
(2) django 使用redis缓存
# 导入测试环境 import os import django os.environ.setdefault("DJANGO_SETTINGS_MODULE", "luffyapi.settings.dev") django.setup() from user import models user = models.User.objects.first() # 测试子自定义序列化类 from rest_framework.serializers import ModelSerializer class UserModelSerializer(ModelSerializer): class Meta: model = models.User fields = ['username', 'password'] user_data = UserModelSerializer(user).data # print(user_data) # 原生redis # import redis # r = redis.Redis() # r.set(user.usernaem,user_data) from django.core.cache import cache # >>> # cache.set() # vdef set(self, name, value, ex=None, px=None, nx=False, xx=False): # cache.get() # cache 是一个高速缓存器 cache.set(user.username,user_data,5) res = cache.get(user.username) print(res, type(res)) # {'username': 'admin', 'password': 'pbkdf2_sha256$1000dcKSQ='} # <class 'dict'> 序列化成字典的格式进行存储 # 如和进行数据的存储 现在是默认存储到mameche 需要进行配置存储到我们redis 数据库的缓存 # cache比原生的redis 的好处: (1)djano 和drf份序列化结构 # (2) 如果是普通的json格式 我们就序列化成json格式的字符串 # (3)如果是图片的我们通过图片格式度成二进制
redis实现数据缓存的接口实列
from rest_framework.views import APIView from .models import User from utils.response import APIResponse import re # 注册逻辑：1.校验手机号是否存在 2.发送验证码 3.完成注册 # 校验手机号 class MobileAPIView(APIView): def post(self, request, *args, **kwargs): mobile = request.data.get('mobile') if not mobile or not re.match(r'^1[3-9]d{9}$', mobile): return APIResponse(1, '数据有误') try: User.objects.get(mobile=mobile) return APIResponse(2, '已注册') except: return APIResponse(0, '未注册') # 发送验证码 from libs import txsms from django.core.cache import cache from settings.const import SMS_EXP, SMS_CACHE_KEY from .thorttles import SMSRateThrottle class SMSAPIView(APIView): # 频率限制 throttle_classes = [SMSRateThrottle] def post(self, request, *args, **kwargs): # 1）拿到前台的手机号 mobile = request.data.get('mobile') if not mobile or not re.match(r'^1[3-9]d{9}$', mobile): return APIResponse(2, '数据有误') # 2）调用txsms生成手机验证码 code = txsms.get_code() # 3）调用txsms发送手机验证码 result = txsms.send_sms(mobile, code, SMS_EXP // 60) # 4）失败反馈信息给前台 if not result: return APIResponse(1, '短信发送失败') # 5）成功服务器缓存手机验证码 - 用缓存存储(方便管理) - redis cache.set(SMS_CACHE_KEY % {'mobile': mobile}, code, SMS_EXP) # 6）反馈成功信息给前台 return APIResponse(0, '短信发送成功') # 注册 from rest_framework.generics import CreateAPIView from . import serializers class RegisterCreateAPIView(CreateAPIView): # queryset = User.objects.filter(is_active=True) serializer_class = serializers.RegisterModelSerializer # 自定义响应结果 def create(self, request, *args, **kwargs): serializer = self.get_serializer(data=request.data) serializer.is_valid(raise_exception=True) # 校验失败就主动抛异常 => 自定义异常结果，配置异常模块 user_obj = serializer.save() # 要自定义入库逻辑，重写create方法 headers = self.get_success_headers(serializer.data) # 响应结果需要格式化，使用序列化类要提供序列化与反序列化两套规则 return APIResponse(0, 'ok', results=serializers.RegisterModelSerializer(user_obj).data, http_status=201, headers=headers ) # 多方式登录 class LoginAPIView(APIView): # 1) 禁用认证与权限组件 authentication_classes = [] permission_classes = [] def post(self, request, *args, **kwargs): # 2) 拿到前台登录信息，交给序列化类，规则：账号用usr传，密码用pwd传 序列化入库 (data=request.data) user_ser = serializers.LoginModelSerializer(data=request.data) # 3) 序列化类校验得到登录用户与token存放在序列化对象中 user_ser.is_valid(raise_exception=True) # 4) 取出登录用户与token返回给前台 反序列化 return APIResponse(token=user_ser.token, results=serializers.LoginModelSerializer(user_ser.user).data) # 手机验证码登录 from rest_framework_jwt.serializers import jwt_payload_handler from rest_framework_jwt.serializers import jwt_encode_handler class LoginMobileAPIView(APIView): authentication_classes = [] permission_classes = [] def post(self, request, *args, **kwargs): # 收机号码这是必填项 mobile = request.data.get('mobile') # 还有验证码 code = request.data.get('code') print(code) # 判断验证是否是有值 if not mobile or not code: return APIResponse(1, '数据有误') # 再判断后端生成的验证码是否一致 后端生成 保存在我们的redis 中 安装 settings 进行配置 old_code = cache.get(SMS_CACHE_KEY % {'mobile': mobile}) # 建立缓存 print(old_code) if code != old_code: # 生成就在redis 中进行缓存 和饿哦们自己输入的是否一样 return APIResponse(1, '验证码错误') try: # 登录成功 user = User.objects.get(mobile=mobile) # 用户生成paylaod payload = jwt_payload_handler(user) # 用载荷生成token 登录成功返回token到前端 需要我们浏览器保存token cookies 进行其他需要登录的操作 token = jwt_encode_handler(payload) return APIResponse(token=token, results=serializers.LoginModelSerializer(user).data) except: return APIResponse(1, '用户不存在')
settings >>>dev开发环境下进行配置
# import logging 测试的话导入一下 # logging.getLogger('django') # 配制redis 缓存是用:需要额外的安装django-redis # 步骤: (1) 将缓存存储位置配置到redis 中 CACHES = { 'default':{ 'BACKEND':'django_redis.cache.RedisCache', 'LOCATION':'redis://127.0.0.1:6379/10', 'OPTIONS':{ 'CLIENT_CLASS':'django_redis.client.DefaultClient', 'CONNECTION_POOL_KWARGS':{'max_connections':100} } } }
---恢复内容结束---