小君大帅比前端开发 2020-07-30

Flask-RESTful、Flask-Cache扩展的使用

flask项目结构如图：

具体的文件内容参考：Flask项目拆分，代码基本上都是一样的！视图函数views包下存放对应的API资源，views/__init__.py

from flask_restful import  Api

from app.views.HelloApi import Hello

api = Api()


def init_api(app):
    api.init_app(app)


api.add_resource(Hello, '/', method=['GET', 'POST'])

views/HelloApi.py

from flask_restful import Resource


class Hello(Resource):

    def get(self):
        return {'msg': ' get success'}

    def post(self):
        return {'msg': 'post success'}

输出格式定制

views/StudentApi.py：

from flask_restful import Resource, marshal_with, fields
from app.models import Student

stu_fields = {
    'id': fields.Integer,
    'name': fields.String,
    'age': fields.Integer,
    # 关联数据库中的age列
    's_age': fields.Integer(attribute='age'),
    # hobby在数据表中不存在，可以设置默认值
    'hobby': fields.String(default='ball'),
    # 获取对应资源的url，absolute设置为绝对路径
    'url': fields.Url('students', absolute=True),
}

result_fields = {
    'msg': fields.String,
    'status': fields.String,
    # 数据嵌套,返回一个list列表
    'data': fields.List(fields.Nested(stu_fields))

}


class Students(Resource):
    @marshal_with(result_fields)
    def get(self):
        data = {
            'msg': 'success',
            'status': '200',
        }
        stu = Student.query.all()
        data['data'] = stu

        return data

views/__init.py

from flask_restful import Api

from app.views.HelloApi import Hello
from app.views.StudentApi import Students

api = Api()


def init_api(app):
    api.init_app(app)


api.add_resource(Hello, '/', methods=['GET', 'POST'])
api.add_resource(Students, '/stu', methods=['GET', 'POST'])

输入参数定制

URL:http://127.0.0.1:5000/grade?grade_id=1 views/GradeApi.py

from flask import request
from flask_restful import Resource, reqparse

# 请求参数转换器
parse = reqparse.RequestParser()
# parse.add_argument('grade_id')

# 设置grade_id类型为int, help参数为报错信息
# parse.add_argument('grade_id', type=int, help='参数错误')

# 设置grade_id参数为必须
parse.add_argument('grade_id', required=True, help='参数缺失')

# 存在多个grade_id,返回类型list
parse.add_argument('grade_id', action='append', required=True, help='参数缺失')

# 从headers获取host
parse.add_argument('host', location='headers')


class Grade(Resource):
    # # 获取URL中参数的方法
    # def get(self):
    #
    #     grade_id = request.args.get('grade_id')
    #     print(grade_id)
    #     print(type(grade_id))
    #
    #     return {'msg': 'success'}

    def get(self):
        parser = parse.parse_args()
        grade_id = parser.get('grade_id')
        print(grade_id)
        print(type(grade_id))

        return {'msg': 'success'}

views/__init__.py

from flask_restful import Api

from app.views.GradeApi import Grade
from app.views.HelloApi import Hello
from app.views.StudentApi import Students

api = Api()


def init_api(app):
    api.init_app(app)


api.add_resource(Hello, '/', methods=['GET', 'POST'])
api.add_resource(Students, '/stu', methods=['GET', 'POST'])
api.add_resource(Grade, '/grade', methods=['GET', 'POST'])

获取、修改、删除数据：views/__init__.py


from flask_restful import Api

from app.views.GradeApi import Grade
from app.views.HelloApi import Hello
from app.views.StudentApi import Students

api = Api()


def init_api(app):
    api.init_app(app)


api.add_resource(Hello, '/', methods=['GET', 'POST'])
api.add_resource(Students, '/stu/<int:id>', methods=['GET', 'POST', 'DELETE'])
api.add_resource(Grade, '/grade', methods=['GET', 'POST'])

views/StudentApi.py

from flask_restful import Resource, marshal_with, fields, reqparse

from app.ext import db
from app.models import Student

stu_fields = {
    'id': fields.Integer,
    'name': fields.String,
    'age': fields.Integer,
    # 关联数据库中的age列
    's_age': fields.Integer(attribute='age'),
    # hobby在数据表中不存在，可以设置默认值
    'hobby': fields.String(default='ball'),
    # 获取对应资源的url，absolute设置为绝对路径
    'url': fields.Url('students', absolute=True),
}

result_fields = {
    'msg': fields.String,
    'status': fields.String,
    # # 数据嵌套,返回一个list列表
    # 'data': fields.List(fields.Nested(stu_fields))
    # 获取1个学生
    'data': fields.Nested(stu_fields)

}

# 请求参数
parse = reqparse.RequestParser()
parse.add_argument('name', required=True, help='must supply name')
parse.add_argument('age', required=True, help='must supply age')


class Students(Resource):

    # 根据id获取数据
    @marshal_with(result_fields)
    def get(self, id=None):
        data = {
            'msg': 'success',
            'status': '200',
        }
        stu = Student.query.get(id)
        if stu:
            data['data'] = stu
        else:
            data['data'] = 'fail'
        return data

    # 修改数据
    @marshal_with(result_fields)
    def post(self, id=None):
        stu = Student.query.get(id)

        parser = parse.parse_args()
        name = parser.get('name')
        age = parser.get('age')

        stu.name = name
        stu.age = age

        db.session.add(stu)
        db.session.commit()
        data = {
            'msg': 'success',
            'status': '200',
            'data': stu
        }
        return data
    
    # 删除数据
    def delete(self, id=None):
        # # 如果没有404
        # stu = Student.query.get_or_404(id)

        stu = Student.query.get(id)
        data = {}

        if stu:
            db.session.delete(stu)
            db.session.commit()

            data['status'] = '204'
            data['msg'] = 'delete success'
            return data
        else:
            data['status'] = '404'
            data['msg'] = 'does not exits'

            return data

接口缓存

app/ext.py

from flask_cache import Cache
from flask_migrate import Migrate
from flask_sqlalchemy import SQLAlchemy

db = SQLAlchemy()
migrate = Migrate()

cache = Cache(config={
    'CACHE_TYPE': 'redis'  # 配置redis
})


def init_ext(app):
    db.init_app(app)
    migrate.init_app(app, db)
    cache.init_app(app)

viwes/StudentApi.py

import time

from flask_restful import Resource, marshal_with, fields, reqparse

from app.ext import db, cache
from app.models import Student

stu_fields = {
    'id': fields.Integer,
    'name': fields.String,
    'age': fields.Integer,
    # 关联数据库中的age列
    's_age': fields.Integer(attribute='age'),
    # hobby在数据表中不存在，可以设置默认值
    'hobby': fields.String(default='ball'),
    # 获取对应资源的url，absolute设置为绝对路径
    'url': fields.Url('students', absolute=True),
}

result_fields = {
    'msg': fields.String,
    'status': fields.String,
    # # 数据嵌套,返回一个list列表
    # 'data': fields.List(fields.Nested(stu_fields))
    # 获取1个学生
    'data': fields.List(fields.Nested(stu_fields))

}

# http://127.0.0.1:5000/stu
class Students(Resource):

    @marshal_with(result_fields)
    @cache.cached(timeout=30)
    def get(self,):
        data = {
            'msg': 'success',
            'status': '200',
        }
        stu = Student.query.all()
        time.sleep(3)
        data['data'] = stu

        return data

需要安装并启动redis，python解释器需要安装reids，flask版本太高不兼容报错，0.12版本最合适。

flask