依赖包如下：

(venv) C:\Users\fengzi\Desktop\flaskProject3>pip list
Package                Version
---------------------- --------
arrow                  1.2.0
cachelib               0.4.1
cffi                   1.14.6
click                  8.0.1
colorama               0.4.4
cryptography           3.4.7
Deprecated             1.2.13
flags                  0.0.1.2
Flask                  2.0.2
Flask-Caching          1.10.1
flask-paginate         0.8.1
flask-redis            0.4.0
Flask-Session          0.4.0
Flask-SQLAlchemy       2.5.1
greenlet               1.1.1
importlib-metadata     4.6.4
itsdangerous           2.0.1
Jinja2                 3.0.1
lxml                   4.6.5
MarkupSafe             2.0.1
mysql-connector-python 8.0.26
pamqp                  2.3.0
pika                   1.2.0
pip                    21.3
protobuf               3.17.3
pycparser              2.20
PyMySQL                1.0.2
python-dateutil        2.8.1
pytils                 0.3
rabbitpy               2.0.1
redis                  4.0.2
setuptools             57.4.0
six                    1.16.0
SQLAlchemy             1.4.23
tools                  0.1.9
typing-extensions      3.10.0.0
Werkzeug               2.0.2
wrapt                  1.13.3
XMind                  1.2.0
xmind2testcase         1.5.0
zipp                   3.5.0

 

 

app.py内容如下：

import json
from flask_paginate import Pagination, get_page_parameter, request
from flask import Flask, render_template
from flask_sqlalchemy import SQLAlchemy
from flask_caching import Cache
import pickle
from model.UserModel import query
import json
config = {
    "DEBUG": True,
    "CACHE_TYPE": "RedisCache",
    "CACHE_DEFAULT_TIMEOUT": 300,
    "CACHE_REDIS_HOST": "xxxxx",
    "CACHE_REDIS_PORT": xxx,
    "CACHE_REDIS_DB": 0,
}
app = Flask(__name__)

app.config.from_mapping(config)
cache = Cache(app)


# @cache.cached(timeout=50)
@app.route('/')
def hello_world():
    return render_template("index.html")


@app.route('/list')
# @cache.cached(timeout=60)
def ShowData():
    pageNum = int(request.args.get("pageNum"))
    pageSize = int(request.args.get("pageSize"))
    info = cache.get(f"info{pageNum}")
    if info:
        print("走的redis查询")
        return info
    data_list = query()
    info = {
        "code": 0,
        "message": "OK",
        "total": len(data_list),
        "data": data_list[(pageNum - 1) * pageSize:pageNum * pageSize]
    }
    cache.set(f"info{pageNum}", info)
    print("走的mysql查询")
    return info

@app.route("/addpolicy")
def AddPolicy():
    return render_template('addpolicy.html')


if __name__ == '__main__':
    app.run(debug=True, )

config.py内容如下

host = 'xxxxxxxx'
port = 3306
username = 'xxx'
password = 'xxx'
DB_URI = "mysql+pymysql://{username}:{password}@{host}:{port}/{db}?charset=utf8".format(username=username,password=password, host=host,port=port, db='testdb')
SQLALCHEMY_DATABASE_URI = DB_URI
SQLALCHEMY_TRACK_MODIFICATIONS = False
SQLALCHEMY_ECHO = True

UserModel.py内容如下：

import json
from sqlalchemy import Column,Integer,String
from sqlalchemy.ext.declarative import declarative_base
from sqlalchemy import create_engine
import config
from sqlalchemy.orm import sessionmaker
engine = create_engine(config.SQLALCHEMY_DATABASE_URI)
Base = declarative_base()

class User(Base):
    __tablename__ = 'users'
    id = Column(Integer,primary_key=True,autoincrement=True)
    name = Column(String(32))
    age = Column(Integer)
    def __init__(self,name,age):
        self.name = name
        self.age = age
    def __str__(self):
        info = {
            "id": self.id,
            "name": self.name,
            "age": self.age
        }
        return json.dumps(info)

def query():
    sm = sessionmaker(bind=engine)
    session = sm()
    all = session.query(User).all()
    data_list = []
    for res in all:
        datas = {
            "id": res.id,
            "name": res.name,
            "age": res.age
        }
        data_list.append(datas)
    return data_list

效果如下：

在第一次访问网站的时候可以看到走的是mysql关系型数据库，当再次访问的时候走的是redis查询了

 

 登录到redis服务器查看结果，可以看到，一共6页内容，缓存了6次，有6个key