Flask中redis的配置与使用

 注意点：

         在__init__.py中需要将redis_store设置成全局变量，这样方便其他文件导入

一、config.py

import logging
import os
from datetime import timedeltafrom redis import StrictRedisclass Config:# 调试信息DEBUG = TrueSECRET_KEY = os.urandom(32)JWT_SECRET = os.urandom(32)JWT_EXPIRY_SECOND = 5# 数据库信息MYSQL_DIALECT = 'mysql'MYSQL_DIRVER = 'pymysql'MYSQL_NAME = 'root'MYSQL_PWD = '123456'MYSQL_HOST = 'localhost'MYSQL_PORT = 3306MYSQL_DB = 'manger'MYSQL_CHARSET = 'utf8mb4'SQLALCHEMY_DATABASE_URI = f'{MYSQL_DIALECT}+{MYSQL_DIRVER}://{MYSQL_NAME}:{MYSQL_PWD}@{MYSQL_HOST}:{MYSQL_PORT}/{MYSQL_DB}?charset={MYSQL_CHARSET}'SQLALCHEMY_TRACK_MODIFICATIONS = True# redis配置REDIS_HOST = '127.0.0.1'REDIS_PORT = 6379# session配置# SESSION_TYPE = 'filesystem'  # 设置session的存储类型# SESSION_REDIS = StrictRedis(host=REDIS_HOST, port=REDIS_PORT)  # 指定session存储的服务器# SESSION_USE_SIGNER = True  # 设置签名存储# PERMANENT_SESSION_LIFETIME = timedelta(days=1)  # 设置签名过期时间# 配置默认的log等级LEVEL_NAME = logging.DEBUG# 开发环境配置信息
class DevelopConfig(Config):pass# 生产（线上）环境配置信息
class ProductConfig(Config):DEBUG = False# 测试环境配置信息
class TestConfig(Config):pass# 提供一个统一的访问入口
config_dict = {"develop": DevelopConfig,"product": ProductConfig,"test": TestConfig
}

二、__init__.py

import logging
from logging.handlers import RotatingFileHandlerfrom flask import Flask
from flask_sqlalchemy import SQLAlchemy
from redis import StrictRedis
from flask_session import Sessionfrom config import Config, config_dictdb = SQLAlchemy()
redis_store = Nonedef create_app(config_name):app = Flask(__name__)# 获取config配置config = config_dict.get(config_name)app.config.from_object(config)# 调用日志方法，记录程序运行信息log_file(config.LEVEL_NAME)# 创建数据库关联对象并关联appdb.init_app(app)# 创建redis对象# 当 decode_responses 设置为 True 时，Redis 返回的字符串数据将会被解码为 Python 字符串类型。这样可以方便地处理 Redis 中存储的文本数据。# 而当 decode_responses 设置为 False（默认值）时，Redis 返回的字符串数据将会以字节字符串（bytes）的形式返回。# 这在处理二进制数据或者需要与其他 Redis 客户端进行交互时可能更为合适global redis_storeredis_store = StrictRedis(host=config.REDIS_HOST, port=config.REDIS_PORT, decode_responses=True)# # 创建session对象# Session(app)# 注册蓝图from manger.user import user_blueapp.register_blueprint(user_blue)return app

三、app.py

from manger import create_app, db
from flask_migrate import Migrate
from manger import models# 传入settings参数，开发版本“develop”，线上版本“product”
app = create_app('develop')# 数据库设置
migrate = Migrate(app, db)@app.route('/')
def hello_world():  # put application's code herereturn 'Hello World!'if __name__ == '__main__':app.run()

四、views中使用，登录

@user_blue.route('/login', methods=['POST'])
def login():username = request.json.get('username')password = request.json.get('password')if not all([username, password]):return JsonResponse.error(msg='缺少必要参数！').to_response()try:user_obj = models.User.query.filter(models.User.name == username).first()except Exception as e:current_app.logger.error(e)if not user_obj.check_password(password):return JsonResponse.error(msg='密码错误！').to_response()token = encode_token(user_obj.id, username, password)# 将token存入redis中redis_store.set('token', token)# 设置过期时间redis_store.expire('token', 20)  data = {"data": user_obj.to_dict(),"token": token}return JsonResponse(code=200, data=data, msg='登录成功！').to_response()