Flask使用redis缓存

flask使用redis数据库来做缓存，我们需要先安装Redis服务,安装python操作redis的客户端库redis，接着安装使用flask-caching

linux系统（以centos为例）安装Redis服务

sudo yum install -y redis

# 启动Redis并设置开机自启
sudo systemctl start redis
sudo systemctl enable redis
# 查看版本，确认安装
redis-server --version  
# 验证安装
redis-cli ping

安装python操作redis的客户端库redis

使用 pip install redis 安装的是 Python 操作 Redis 的客户端库，这与 Redis 服务器本身是分开的。

简单来说：

pip install redis 安装的是让 Python 代码能够连接和操作 Redis 服务器的工具库
但要使用 Redis 服务，还需要单独安装并启动 Redis 服务器。

安装客户端库后，你的 Python 代码，就可以通过 import redis 来使用这个库，连接到本地启动的 Redis 服务器了。

安装redis:

pip install redis

flask-caching介绍

Flask-Caching 是 Flask 框架的一个扩展，用于在 Flask 应用中实现缓存功能，通过存储计算结果或频繁访问的数据来提高应用性能，减少不必要的计算和数据库查询。

主要功能

多种缓存后端支持：
- 内存缓存（适合开发环境）
- 文件系统缓存
- Redis、Memcached 等分布式缓存（适合生产环境）
- 其他如 MongoDB、SQLAlchemy 等
灵活的缓存方式：
- 装饰器方式：缓存视图函数或普通函数的返回值
- 低级 API：手动控制缓存的设置、获取和删除
- 模板缓存：缓存 Jinja2 模板片段
缓存配置选项：
- 缓存过期时间设置
- 缓存键的前缀和生成规则
- 不同视图 / 函数可设置不同缓存策略

基本使用步骤

安装

pip install flask-caching

初始化配置：

from flask import Flask
from flask_caching import Cache

app = Flask(__name__)

# 配置缓存
app.config['CACHE_TYPE'] ='redis'  # 指定缓存类型为Redis
app.config['CACHE_REDIS_HOST'] = 'localhost'  # Redis主机地址，根据实际情况修改
app.config['CACHE_REDIS_PORT'] = 6379  # Redis端口，默认是6379，根据实际情况修改
app.config['CACHE_REDIS_DB'] = 0  # Redis数据库编号，默认为0，根据实际情况修改
app.config['CACHE_REDIS_PASSWORD'] = None  # 如果Redis设置了密码，这里填写密码

# 初始化缓存（若创建了扩展文件，则在扩展文件中初始化）
cache = Cache(app)

若使用了工厂函数，则这样配置：

from .exts import  cache
config = {
    'CACHE_TYPE': 'redis',
    'CACHE_REDIS_HOST': '127.0.1',
    'CACHE_REDIS_PORT': 6379
}

def create_app():
    app = Flask(__name__, template_folder='../templates', static_folder='../static')
    app.config.from_object(Config)
    ##其他配置
    cache.init_app(app=app, config=config)
    return app

缓存视图函数

@app.route('/api/data')
@cache.cached(timeout=60)  # 缓存 60 秒
def get_data():
    # 模拟耗时操作（如数据库查询）
    return {"data": "需要缓存的内容"}

缓存普通函数

@cache.cached(key_prefix='my_function', timeout=120)
def expensive_function(param):
    # 耗时计算
    return result

模板缓存

{% cache 500, 'template_block' %}  # 缓存 500 秒
    <div>需要缓存的模板内容</div>
{% endcache %}

手动操作缓存

# 设置缓存
cache.set('key', 'value', timeout=60)

# 获取缓存
value = cache.get('key')

# 删除缓存
cache.delete('key')

# 清空所有缓存
cache.clear()

适用场景

频繁访问但不常变化的数据（如分类列表、配置信息）
计算密集型操作的结果
数据库查询结果缓存
API 响应缓存

使用 Flask-Caching 可以显著提升应用的响应速度，减轻服务器和数据库负担，尤其适合流量较大的应用。选择合适的缓存后端和合理的缓存策略是关键。

下面展示邮箱验证码案例：

获取到验证码并将它设置在redis缓存中

@user_bp.route('/captcha/email/')
def get_email_captcha():
    # /captcha/email/<email>
    # /captcha/email/?email=xxx@qq.com
    email = request.args.get('email')
    source = string.digits * 4
    captcha = random.sample(source, 4)
    captcha = ''.join(captcha)
    message = Message(subject='验证码', recipients=[email], body=f'您的验证码是{captcha}')
    mail.send(message)
    cache.set(email, captcha, timeout=300)
    # RESTfulAPI
    return jsonify({'code': 200, 'message': '', 'data': None})

获取并进行验证：

class RegisterForm(wtforms.Form):
    username = wtforms.StringField(validators=[Length(min=1, max=20, message="用户名格式错误！")])
    email = wtforms.StringField(validators=[Email(message="邮箱格式错误！")])
    captcha = wtforms.StringField(validators=[Length(min=4, max=4, message="验证码格式错误！请输入四位数字验证码！")])
    password = wtforms.StringField(validators=[Length(min=6, max=16, message="密码格式错误！最小6位，最大16位！")])
    password_confirm = wtforms.StringField(validators=[EqualTo('password', message="两次密码输入不一致")])

    # 自定义验证:1.邮箱是否被注册2.验证码是否正确
    def validate_email(self, field):
        # field.data,可以获取到用户输入到该字段的值
        email = field.data
        user = User.query.filter_by(email=email).first()
        if user:
            raise wtforms.ValidationError(message="该邮箱已经被注册！")

    def validate_captcha(self, field):
        captcha = field.data
        email = self.email.data
        captcha_redis = cache.get(email)
        if captcha != captcha_redis:
            raise wtforms.ValidationError(message="验证码错误！")

返回文章列表

文章目录