flask-restful 组件

目录

• flask-restful 组件
  • 一、 简介
    • 1、 介绍
    • 2、 rest 特性
    • 3、 rest 理念
    • 4、 环境配置
    • 5、 最小的API
  • 二、 请求解析
    • 1、 基本参数
    • 2、 必须参数
    • 3、 多值参数
    • 4、 参数转移
    • 5、 参数位置
    • 6、 多个位置
    • 7、 继承解析
  • 三、 输出字段
    • 1、 基本用法
    • 2、 重命名属性
    • 3、 默认值
    • 4、 自定义字段
    • 5、 url字段
    • 6、 复杂结构
    • 7、 列表字段
  • 四、 扩展知识
    • 1、 响应数据
    • 2、 参数验证
    • 3、 资源方法装饰器
    • 4、 错误处理器
  • 五、 服务案例
    • 1、 环境
    • 2、 模型映射
    • 3、 创建路由
    • 4、 创建restful服务
    • 5、 初始化app

flask-restful 组件 一、 简介 1、 介绍

Flask-restful 是一个 Flask 扩展，它添加了快速构建 REST APIs 的支持。它当然也是一个能够跟你现有的ORM/库协同工作的轻量级的扩展。Flask-RESTful 鼓励以最小设置的最佳实践。如果你熟悉 Flask 的话，Flask-restful 应该很容易上手。

关于flask的使用，参考我之前的博客：https://blog.csdn.net/qq_62789540/article/details/124781918

关于flask的组件，参考我之前的文章：https://blog.csdn.net/qq_62789540/article/details/125563241

flask-restful官方文档：https://flask-restful.readthedocs.io/en/latest/quickstart.html

flask-restful中文文档：http://www.pythondoc.com/Flask-RESTful/index.html

2、 rest 特性

• Client-Server：服务器端与客户端分离。
• Stateless（无状态）：每次客户端请求必需包含完整的信息，换句话说，每一次请求都是独立的。
• Cacheable（可缓存）：服务器端必需指定哪些请求是可以缓存的。
• Layered System（分层结构）：服务器端与客户端通讯必需标准化，服务器的变更并不会影响客户端。
• Uniform Interface（统一接口）：客户端与服务器端的通讯方法必需是统一的。
• Code on demand（按需执行代码？）：服务器端可以在上下文中执行代码或者脚本？

3、 rest 理念

REST架构就是为了HTTP协议设计的。RESTful web services的核心概念是管理资源。资源是由URIs来表示，客户端使用HTTP当中的'POST, OPTIONS, GET, PUT, DELETE'等方法发送请求到服务器，改变相应的资源状态。

HTTP请求方法通常也十分合适去描述操作资源的动作：

HTTP方法 动作 例子 GET 获取资源信息 http://example.com/api/orders（检索订单清单） GET 获取资源信息 http://example.com/api/orders/123（检索订单 #123） POST 创建一个次的资源 http://example.com/api/orders（使用带数据的请求，创建一个新的订单） PUT 更新一个资源 http://example.com/api/orders/123（使用带数据的请求，更新#123订单） DELETE 删除一个资源 http://example.com/api/orders/123删除订单#123

REST请求并不需要特定的数据格式，通常使用JSON作为请求体，或者URL的查询参数的一部份。

4、 环境配置

安装：

pip install flask
pip install flask-restx

5、 最小的API

# ！/usr/bin/python3
# -*- coding: UTF-8 -*-
__author__ = "A.L.Kun"
__file__ = "app3.py"
__time__ = "2022/7/8 9:30"

from flask import Flask, url_for
from flask_restx import Resource, Api

app = Flask(__name__)
api = Api(app)
"""
也可以这样
api = Api()
app = Flask(__name__)
api.init_app(app)
"""


# 这里可以使用类装饰器，注册路由，其可以做add_resource能做的事情
# @api.route("/", "/index", endpoint="hello")
class HelloWorld(Resource):
    def get(self):
        # print(url_for("hello"))
        return {
            "name": "hello world"
        }

    def put(self):
        pass

    def post(self):
        pass

    def delete(self):
        pass


api.add_resource(HelloWorld, '/', "/hello", endpoint="hello")  # 添加路由
if __name__ == '__main__':
    app.run("0.0.0.0", 5000, debug=True)

二、 请求解析

Flask-RESTful的请求解析接口是模仿argparse接口。它设计成提供简单并且同一的访问Flask中flask.request 对象里的任何变量的入口

1、 基本参数

from flask_restful import reqparse

parser = reqparse.RequestParser()  # 创建一个参数接收器
parser.add_argument('rate', type=int, help='Rate cannot be converted')  # 添加基本参数
parser.add_argument('name', type=str)
args = parser.parse_args()  # 解析参数

如果你指定了 help 参数的值，在解析的时候当类型错误被触发的时候，它将会被作为错误信息给呈现出来。如果你没有指定 help 信息的话，默认行为是返回类型错误本身的信息。

默认下，arguments 不是 必须的。另外，在请求中提供的参数不属于 RequestParser 的一部分的话将会被忽略。

另请注意：在请求解析中声明的参数如果没有在请求本身设置的话将默认为 None。

2、 必须参数

要求一个值传递的参数，只需要添加required=True来调用add_argument()

parser.add_argument('name', type=str, required=True,
help="Name cannot be blank!")

3、 多值参数

如果你要接受一个键有多个值的话，你可以传入 action='append'

parser.add_argument('name', type=str, action='append')

这样，查询的话：

curl http://api.example.com -d "Name=bob" -d "Name=sue" -d "Name=joe"

4、 参数转移

如果由于某些原因，你想要以不同的名称存储你的参数，一旦它被解析的时候，你可以使用dest=arg

parser.add_argument('name', type=str, dest='public_name')

args = parser.parse_args()
print(args['public_name'])

5、 参数位置

默认下，RequestParser 试着从 flask.Request.values，以及 flask.Request.json 解析值。

在 add_argument() 中使用 location 参数可以指定解析参数的位置。flask.Request 中任何变量都能被使用。例如：

# Look only in the POST body
parser.add_argument('name', type=int, location='form')

# Look only in the querystring
parser.add_argument('PageSize', type=int, location='args')

# From the request headers
parser.add_argument('User-Agent', type=str, location='headers')

# From http cookies
parser.add_argument('session_id', type=str, location='cookies')

# From file uploads
parser.add_argument('picture', type=werkzeug.datastructures.FileStorage, location='files')

常见参数位置：

6、 多个位置

通过传入一个列表到locaion中可以指定多个参数位置：

parser.add_argument('text', location=['headers', 'values'])

列表中最后一个优先出现在结果集中。（例如：location=[‘headers’, ‘values’]，解析后 ‘values’ 的结果会在 ‘headers’ 前面）

7、 继承解析

往往你会为你编写的每个资源编写不同的解析器。这样做的问题就是如果解析器具有共同的参数。不是重写，你可以编写一个包含所有共享参数的父解析器接着使用 copy() 扩充它。你也可以使用 replace_argument() 覆盖父级的任何参数，或者使用 remove_argument() 完全删除参数。

from flask_restful import RequestParser

parser = RequestParser()
parser.add_argument('foo', type=int)

parser_copy = parser.copy()
parser_copy.add_argument('bar', type=int)

# parser_copy has both 'foo' and 'bar'

parser_copy.replace_argument('foo', type=str, required=True, location='json')
# 'foo' is now a required str located in json, not an int as defined
#  by original parser

parser_copy.remove_argument('foo')
# parser_copy no longer has 'foo' argument

三、 输出字段 1、 基本用法

你可以定义一个字典或者 fields 的 OrderedDict 类型，OrderedDict 类型是指键名是要呈现的对象的属性或键的名称，键值是一个类，该类格式化和返回的该字段的值

from flask.ext.restful import Resource, fields, marshal_with

resource_fields = {
    'name': fields.String,  # 创建字符串字段
    'address': fields.String,  # 字符串字段
    'date_updated': fields.DateTime(dt_format='rfc822'),  # 格式为RFC 822 的时间字符串
}

class Todo(Resource):
    @marshal_with(resource_fields, envelope='resource')  # 标准化返回的资源
    def get(self, **kwargs):
        return db_get_todo()  # Some function that queries the db

这个例子假设你有一个自定义的数据库对象（todo），它具有属性：name， address， 以及 date_updated。该对象上任何其它的属性可以被认为是私有的不会在输出中呈现出来。一个可选的 envelope 关键字参数被指定为封装结果输出。

装饰器 marshal_with 是真正接受你的数据对象并且过滤字段。marshal_with 能够在单个对象，字典，或者列表对象上工作。

注意：marshal_with 是一个很便捷的装饰器，在功能上等效于如下的 return marshal(db_get_todo(), resource_fields), 200。这个明确的表达式能用于返回 200 以及其它的 HTTP 状态码作为成功响应（错误响应见 abort）。

2、 重命名属性

很多时候你面向公众的字段名称是不同于内部的属性名。使用 attribute 可以配置这种映射。

fields = {
    'name': fields.String(attribute='private_name'),
    'name_': fields.String(attribute=lambda x: x._private_name),
    'address': fields.String,
}

3、 默认值

如果由于某种原因你的数据对象中并没有你定义的字段列表中的属性，你可以指定一个默认值而不是返回 None。

fields = {
    'name': fields.String(default='Anonymous User'),
    'address': fields.String,
}

4、 自定义字段

有时候你有你自己定义格式的需求。你可以继承 fields.Raw 类并且实现格式化函数。当一个属性存储多条信息的时候是特别有用的。例如，一个位域（bit-field）各位代表不同的值。你可以使用 fields 复用一个单一的属性到多个输出值（一个属性在不同情况下输出不同的结果）。

这个例子假设在 flags 属性的第一位标志着一个“正常”或者“迫切”项，第二位标志着“读”与“未读”。这些项可能很容易存储在一个位字段，但是可读性不高。转换它们使得具有良好的可读性是很容易的。

class UrgentItem(fields.Raw):
    def format(self, value):
        return "Urgent" if value & 0x01 else "Normal"

class UnreadItem(fields.Raw):
    def format(self, value):
        return "Unread" if value & 0x02 else "Read"

class RandomNumber(fields.Raw):
    def output(self, key, obj):
        return random.random()

fields = {
    'name': fields.String,
    'priority': UrgentItem(attribute='flags'),
    'status': UnreadItem(attribute='flags'),
    'random': RandomNumber,
}

自定义输出字段让你无需直接修改内部对象执行自己的输出格式。所有你必须做的就是继承 Raw 并且实现 format() 方法

5、 url字段

Flask-RESTful 包含一个特别的字段，fields.Url，即为所请求的资源合成一个 uri。这也是一个好示例，它展示了如何添加并不真正在你的数据对象中存在的数据到你的响应中。

fields = {
    'uri': fields.Url('todo_resource', absolute=True),
    'uri_': fields.Url('todo_resource'),
    'https_uri': fields.Url('todo_resource', absolute=True, scheme='https')
}

默认情况下，fields.Url 返回一个相对的 uri。为了生成包含协议（scheme），主机名以及端口的绝对 uri，需要在字段声明的时候传入 absolute=True。传入 scheme 关键字参数可以覆盖默认的协议（scheme）:

6、 复杂结构

你可以有一个扁平的结构，marshal_with 将会把它转变为一个嵌套结构

>>> from flask.ext.restful import fields, marshal
>>> import json
>>>
>>> resource_fields = {'name': fields.String}
>>> resource_fields['address'] = {}
>>> resource_fields['address']['line 1'] = fields.String(attribute='addr1')
>>> resource_fields['address']['line 2'] = fields.String(attribute='addr2')
>>> resource_fields['address']['city'] = fields.String
>>> resource_fields['address']['state'] = fields.String
>>> resource_fields['address']['zip'] = fields.String
>>> data = {'name': 'bob', 'addr1': '123 fake street', 'addr2': '', 'city': 'New York', 'state': 'NY', 'zip': '10468'}
>>> json.dumps(marshal(data, resource_fields))
'{"name": "bob", "address": {"line 1": "123 fake street", "line 2": "", "state": "NY", "zip": "10468", "city": "New York"}}'

注意：address 字段并不真正地存在于数据对象中，但是任何一个子字段（sub-fields）可以直接地访问对象的属性，就像没有嵌套一样。

7、 列表字段

你也可以把字段解组（unmarshal）成列表

>>> from flask.ext.restful import fields, marshal
>>> import json
>>>
>>> resource_fields = {'name': fields.String, 'first_names': fields.List(fields.String)}
>>> data = {'name': 'Bougnazal', 'first_names' : ['Emile', 'Raoul']}
>>> json.dumps(marshal(data, resource_fields))
>>> '{"first_names": ["Emile", "Raoul"], "name": "Bougnazal"}'

四、 扩展知识 1、 响应数据

开箱即用，Flask-RESTful 仅配置为支持 JSON。我们做出这个决定是为了给 API 维护者完全控制 API 格式支持，因此一年来的路上，你不必支持那些使用 API 且用 CSV 表示的人们，甚至你都不知道他们的存在。要添加其它的 mediatypes 到你的 API 中，你需要在 Api 对象中声明你支持的表示。

为了支持其它的表示（像 XML,CSV,HTML），你可以使用 representation() 装饰器。你需要在你的 API 中引用它。

app = Flask(__name__)
api = restful.Api(app)

@api.representation('application/json')
def output_json(data, code, headers=None):
    resp = make_response(json.dumps(data), code)
    resp.headers.extend(headers or {})
    return resp

@api.representation('text/csv')
def output_csv(data, code, headers=None):
    pass
    # implement csv output!
   
"""
class Api(restful.Api):
    def __init__(self, *args, **kwargs):
        super(Api, self).__init__(*args, **kwargs)
        self.representations = {
            'application/xml': output_xml,
            'text/html': output_html,
            'text/csv': output_csv,
            'application/json': output_json,
        }
"""

这些输出函数有三个参数： data，code，以及 headers

data 是你从你的资源方法返回的对象，code 是预计的 HTTP 状态码，headers 是设置在响应中任意的 HTTP 头。你的输出函数应该返回一个 Flask 响应对象

2、 参数验证

对于解析参数，你可能要执行自定义验证。创建你自己的输入类型让你轻松地扩展请求解析。

def odd_number(value):
    if value % 2 == 0:
        raise ValueError("Value is not odd")

    return value

def odd_number(value, name):
    """请求解析器在你想要在错误消息中引用名称的情况下将也会允许你访问参数的名称。"""
    if value % 2 == 0:
        raise ValueError("The parameter '{}' is not odd. You gave us the value: {}".format(name, value))

    return value

def task_status(value):
    """你还可以将公开的参数转换为内部表示"""
    statuses = [u"init", u"in-progress", u"completed"]
    return statuses.index(value)

parser = reqparse.RequestParser()
parser.add_argument('OddNumber', type=odd_number)
parser.add_argument('Status', type=task_status)
args = parser.parse_args()

3、 资源方法装饰器

Resource() 有一个叫做 method_decorators 的属性。你可以继承 Resource 并且添加你自己的装饰器，该装饰器将会被添加到资源里面所有 method 函数。举例来说，如果你想要为每一个请求建立自定义认证。

def authenticate(func):
    @wraps(func)
    def wrapper(*args, **kwargs):
        if not getattr(func, 'authenticated', True):
            return func(*args, **kwargs)

        acct = basic_authentication()  # custom account lookup function

        if acct:
            return func(*args, **kwargs)

        restful.abort(401)
    return wrapper


class Resource(restful.Resource):
    method_decorators = [authenticate]   # applies to all inherited resources

由于 Flask-RESTful Resources 实际上是 Flask 视图对象，你也可以使用标准的 flask 视图装饰器。

4、 错误处理器

错误处理是一个很棘手的问题。你的 Flask 应用可能身兼数职，然而你要以正确的内容类型以及错误语法处理所有的 Flask-RESTful 错误。

Flask-RESTful 在 Flask-RESTful 路由上发生任何一个 400 或者 500 错误的时候调用 handle_error() 函数，不会干扰到其它的路由。你可能需要你的应用程序在 404 Not Found 错误上返回一个携带正确媒体类型（介质类型）的错误信息；在这种情况下，使用 Api 构造函数的 catch_all_404s 参数。

app = Flask(__name__)
api = flask_restful.Api(app, catch_all_404s=True)  
from flask import got_request_exception  # 导入信号

def log_exception(sender, exception, **extra):
    """ Log an exception to our logging framework """
    sender.logger.debug('Got exception during processing: %s', exception)

got_request_exception.connect(log_exception, app)

Flask-RESTful 会处理除了自己路由上的错误还有应用程序上所有的 404 错误。

有时候你想在发生错误的时候做一些特别的东西 - 记录到文件，发送邮件，等等。使用 got_request_exception() 方法把自定义错误处理加入到异常。

自定义错误消息：

在一个请求期间遇到某些错误的时候，你可能想返回一个特定的消息以及/或者状态码。你可以告诉 Flask-RESTful 你要如何处理每一个错误/异常，因此你不必在你的 API 代码中编写 try/except 代码块。

errors = {
    'UserAlreadyExistsError': {
        'message': "A user with that username already exists.",
        'status': 409,
    },
    'ResourceDoesNotExist': {
        'message': "A resource with that ID no longer exists.",
        'status': 410,  # 状态码，默认是500
        'extra': "Any extra information you want.",
    },
}
app = Flask(__name__)
api = flask_restful.Api(app, errors=errors)

五、 服务案例 1、 环境

然后，初始化我们的项目

# ！/usr/bin/python3
# -*- coding: UTF-8 -*-
__author__ = "A.L.Kun"
__file__ = "app.py"
__time__ = "2022/7/9 21:21"
from flaskApp import create_app, db
from flask_script import Manager
from flask_migrate import Migrate

app = create_app()
manage = Manager(app)
Migrate(app, db)

if __name__ == '__main__':
    manage.run()

配置文件：

# ！/usr/bin/python3
# -*- coding: UTF-8 -*-
__author__ = "A.L.Kun"
__file__ = "settings.py"
__time__ = "2022/7/9 19:40"


class Basic():
    """编写配置文件"""
    DEBUG = True
    # 连接数据的URI
    DB_URI = "sqlite:///app.db"
    SQLALCHEMY_DATABASE_URI = DB_URI  # 使用sqlachemy

    SQLALCHEMY_TRACK_MODIFICATIONS = True

    SWAGGER_TITLE = "API"
    SWAGGER_DESC = "API接口"
    # 地址，必须带上端口号
    SWAGGER_HOST = "localhost:5000"


class Product():
    pass

2、 模型映射

# ！/usr/bin/python3
# -*- coding: UTF-8 -*-
__author__ = "A.L.Kun"
__file__ = "models.py"
__time__ = "2022/7/9 21:32"

from . import db


class Foo(db.Model):
    """
    模型，将映射到数据库表中
    """
    def __init__(self, id, name, age):
        self.id = id
        self.name = name
        self.age = age

    __tablename__ = 'foo'
    # 主键ID
    id = db.Column(db.INTEGER, primary_key=True, autoincrement=True)
    # 名字
    name = db.Column(db.String(100), nullable=False)
    # 年龄
    age = db.Column(db.INTEGER)

然后在终端运行：

flask db init
flask db migrate
flask db upgrade

3、 创建路由

# ！/usr/bin/python3
# -*- coding: UTF-8 -*-
__author__ = "A.L.Kun"
__file__ = "test.py"
__time__ = "2022/7/9 21:23"

from flask import Blueprint, url_for, redirect
from flaskApp.models import Foo
from flaskApp import db

te = Blueprint("te", __name__)

from . import api_te


@te.route("/index")
def index():
    """则，这里只要关注返回页面的请求，返回数据交给restful去处理"""
    return redirect(url_for("te.operateData"))  # 这里就只是进行一个简单的模拟，进行重定向访问，实际开发中可以使用Ajax异步请求


@te.route("/add")
def search():
    """这里为了方便，我们使用add路由来添加数据"""
    db.session.add(Foo(1, "李华", 23))
    db.session.commit()
    db.session.remove()
    return "success"

4、 创建restful服务

# ！/usr/bin/python3
# -*- coding: UTF-8 -*-
__author__ = "A.L.Kun"
__file__ = "api_te.py"
__time__ = "2022/7/9 21:52"

from flask import request, jsonify
from .test import te
from flask_restful import Api, Resource, fields, marshal_with, reqparse
from flaskApp import db
from flaskApp.models import Foo
from flaskApp.tools import HttpCode

api = Api(te)


def restful_result(code, message, data):
    return jsonify({"code": code, "message": message, "data": data or {}})


def success(message="", data=None):
    """
    正确返回
    :return:
    """
    return restful_result(code=HttpCode.ok, message=message, data=data)


class FooListApi(Resource):
    # 定义要返回的字段
    resource_fields = {
        'id': fields.Integer,
        'name': fields.String,
        'age': fields.String
    }

    # 装饰器，定义返回数据
    @marshal_with(resource_fields)
    def get(self):
        """
        返回所有记录
        :return:
        """
        parser = reqparse.RequestParser()
        parser.add_argument('name', type=str, location='args', help='必须输入姓名')  # 注意，如果添加了parser参数解析，则，返回数据需要是JSON数据
        arg = parser.parse_args()
        print(arg)
        # http://127.0.0.1:5000/api/v1/foo?name=1&a=2  终端输出: {'name': '1'}

        # 查询数据库
        foos = db.session.query(Foo).all()
        return {
            "data": foos,
            "status": HttpCode.ok
        }

    def post(self):
        """
        创建一条记录
        :return:
        """
        # 参数
        params = request.get_json()
        name = params.get("name")
        age = params.get("age")
        # 构建一个模型
        foo = Foo(name=name, age=age)

        # 加入到数据库
        db.session.add(foo)
        db.session.commit()

        return success("新增一条记录成功！")


# 所有记录
api.add_resource(FooListApi, '/api/v1/foo', endpoint="operateData")  # 返回数据

5、 初始化app

# ！/usr/bin/python3
# -*- coding: UTF-8 -*-
__author__ = "A.L.Kun"
__file__ = "__init__.py"
__time__ = "2022/7/9 21:20"

from flask import Flask
from . import settings
from flask_sqlalchemy import SQLAlchemy

db = SQLAlchemy()

from .views.test import te  # 必须要在db创建后导入


def create_app():
    app = Flask(__name__)
    app.register_blueprint(te)
    app.config.from_object(settings.Basic)
    db.init_app(app)
    return app

完整项目在：https://github.com/liuzhongkun1/flask_/tree/main/flaskRestful