Flask Restful

Flask
Flask
发布日期:   2024-04-11
更新日期:   2024-10-26
文章字数:   3.2k
阅读时长:   16 分
阅读次数:  

Flask Restful

1.安装

在Flask 中,为了写出更优雅的API ,提供了一个插件FLask Restful .

$ pip install flask_restful

官网

2.基本应用

一个简单的API 类似如下:

#!/usr/bin/env python
# coding=utf-8

from flask import Flask
from flask_restful import Resource,Api

app = Flask(__name__)

# 1.Api绑定app
api = Api(app)

# 定义一个GET/Post方法
class HelloWorld(Resource):
 def get(self):
     return {'hello':'world'}

# 注册URL 资源
api.add_resource(HelloWorld,'/api/hello/')

@app.route('/')
def hello_world():
 return 'Hello World!'


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

测试联通性:

❯ curl http://127.0.0.1:5000/api/hello/
{
 "hello": "world"
}

3.Resource routing 资源路由

Flask_Restful提供的最主要的基础就是资源Resource 类.它是构建在Flask可拔插视图 之上的,利用它可以很容易的访问多个HTTP 方法.

>>> from flask_restful import Resource
>>> help(Resource)
class Resource(MethodView):
# Resource继承自MethodView,这意味者它可以重写 get/post等方法
from flask import Flask,request
from flask_restful import Resource,Api


app = Flask(__name__)

# 1. Api初始化app
api = Api(app)

todos = {}

# 2.重写 Get/Post方法
class TodoSimple(Resource):

 def get(self,todo_id):
     '''重写 get 方法'''
     return {todo_id:todos[todo_id]}

 def post(self,todo_id):
     '''重写 post 方法'''
     todos[todo_id] = request.form.get('data')
     return {todo_id:todos[todo_id]}

# 3.注册资源
# 访问URL的变量部分,可以使用 <converter:variable_name>
# 此部分可以写成
# api.add_resource(TodoSimple,'/api/todos/<todo_id>')
api.add_resource(TodoSimple,'/api/todos/<string:todo_id>')


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

访问:

❯ curl http://127.0.0.1:5000/api/todos/todo1 -d "data=test" -X POST
{
 "todo1": "test"
}
❯ curl http://127.0.0.1:5000/api/todos/todo1
{
 "todo1": "test"
}
❯ curl http://127.0.0.1:5000/api/todos/todo2 -d "data=test2" -X POST
{
 "todo2": "test2"
}
❯ curl http://127.0.0.1:5000/api/todos/todo2
{
 "todo2": "test2"
}

add_resource ,可以添加url 中的变量参数,以及多个url .

4.endpoint

可以在API 中指定多个URL ,他们指定在Api.add_resource() 方法中.每个URL 都能访问Resource .

api.add_resource(HelloWorld, '/','/hello')
# 127.0.0.1:5000
# 127.0.0.1:5000/hello 都可以访问到 HelloWold类中定义的内容

还可以为资源方法指定endpoint

api.add_resource(TodoSimple,'/api/todos/<string:todo_id>',endpoint='todos')

# 1. Api初始化app
api = Api(app)

todos = {}

# 2.重写 Get/Post方法
class TodoSimple(Resource):

 def get(self,todo_id=None):
     '''重写 get 方法'''
     return {todo_id:todos[todo_id]}

 def post(self,todo_id=None):
     '''重写 post 方法'''
     todos[todo_id] = request.form.get('data')
     return {todo_id:todos[todo_id]}

# 3.注册资源
# 访问URL的变量部分,可以使用 <converter:variable_name>
# 此部分可以写成
# api.add_resource(TodoSimple,'/api/todos/<todo_id>')
api.add_resource(TodoSimple,'/api/todos/<string:todo_id>',endpoint='todos')

with app.test_request_context():
 print(url_for('todos',todo_id='test3'))


if __name__ == '__main__':
 app.run(debug=True)
  • endpoint是用来给url_for反转的时候指定的.如果不写endpoint,那么将会使用视图的名字的小写作为endpoint.

以上如果不写endpoint

with app.test_request_context():
    print(url_for('todosimple',todo_id='test3'))

5.Argument Parsing 参数解析

Flask_Restful 提供了简单的表单验证功能,类似与WTForms ,但是非常的简陋.它使用了一个argparse 库.

from flask_restful import reqparse

parse = reqparse.RequestParser()
parse.add_argument('rate', type=int, help='this is a test')
args = parse.parse_args()  # 返回一个字典对象

parse.add_argument() 常用参数有

  • default='value' :指定默认值,如果前端没有传递值,就使用默认值.
  • required=True :要求前端必须传递值,默认为False .
  • type=str :指定数据类型,要求前端必须传递规定的数据类型,可以使用python 字典的数据类型,也可以指定flask_restful.inputs 模块下的值,比如:
    • falsk_restful.inputs.url:验证URL
    • falsk_restful.inputs.boolean:验证布尔值,可以是True/False 或0\1
    • falsk_restful.inputs.date :验证日期格式是YYYY-mm-dd
    • falsk_restful.inputs.reqex():验证正则表达式
  • choices=[] :指定一个列表,前端的值必须是列表中的值.
  • help='xxx‘: 指定出错的一个帮助信息.
  • trim=True : 指定去除前端返回值的空格.默认是False.
from flask import Flask,request
from flask_restful import Api,Resource,reqparse,inputs

app = Flask(__name__)
app.config.update({'DEBUG':True})

# 1.Api绑定app
api = Api(app)

# 2.重载get/post 方法

class RegiseterView(Resource):

    def get(self):
        return {'login':'???'}

    def post(self):
        '''验证的输入'''
        parse = reqparse.RequestParser()
        # 去除前端返回值的空格
        parse.add_argument('username',type=str,help='用户名不正确',trim=True)
        parse.add_argument('password',type=str,help='密码不正确',trim=True)
        # 使用type=int,使用默认值
        parse.add_argument('age',type=int,help='年龄必须是整数',trim=True,default=18)
        # 使用inputs.date 数据类型
        parse.add_argument('birthday',type=inputs.date,help='日期不正确',trim=True)
        # 使用inputs.regex
        parse.add_argument('phone',type=inputs.regex(r'1[3789]\d{9}'),help='手机号码不正确',trim=True)
        # 使用inputs.boolean
        parse.add_argument('gender',type=str,choices=['male','female'],help='性别不正确',trim=True)

        # 打印
        args = parse.parse_args()
        print(args)
        return args

# 3.注册路由
api.add_resource(RegiseterView,'/api/register/',endpoint='register')


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

访问并得到

❯ curl http://127.0.0.1:5000/api/register/ -d "username=Jack&password=aaa&age=19&phone=18618609966&gender=male" -X POST
{
    "username": "Jack",
    "password": "aaa",
    "age": 19,
    "birthday": null,
    "phone": "18618609966",
    "gender": "male"
}

6.格式化字段

Flask_Restful 提供了一种简单的方法来控制在HTTP Response 中实际呈现的数据.可以简单的返回一个字典类型的数据,也可以使用flask_restful.fields 模块,这个模块允许在资源中使用所需要的任何对象(ORM自定义类等等),它还可以格式化或过滤HTTP Response ,而不必担心暴露内部数据结构.

marshal_with装饰器,会返回一个OrderDict对象

>>> from flask_restful import fields, marshal_with
>>> mfields = { 'a': fields.Raw }
>>> @marshal_with(mfields)
... def get():
...     return { 'a': 100, 'b': 'foo' }
...
...
>>> get()
OrderedDict([('a', 100)])

1.基本用法

可以定义一个fields的OrderedDict 字典,键指向属性名称或要呈现在对象上的键,值是格式化并返回该字段的类.类似如下:

from flask_restful import Resource,fields,marshal_with,Api
from flask import Flask
from datetime import datetime
# pytz是一个时区的datetime object
import pytz

app = Flask(__name__)

# 1.Api绑定app
api = Api(app)

class Register:
 def __init__(self,name):
     self.name = name
     self.date = datetime.now(tz=pytz.timezone('Asia/Shanghai'))

# 2.重载Get/Post方法
class Todo(Resource):

 resource_fields = {
     'name':fields.String,
     'date': fields.DateTime(dt_format='rfc822')
 }

 @marshal_with(resource_fields)
 def get(self,name):
     print(name)
     return Register(name),200  # 可以返回状态码

# 3.注册路由/资源
api.add_resource(Todo,'/api/todo/<name>/',endpoint='todo')

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

访问并得到:

❯ curl http://127.0.0.1:5000/api/todo/Jack/
{
 "name": "Jack",
 "date": "Wed, 11 Dec 2015 15:07:36 -0000"
}

@marshel_with 是一个装饰器,能真正接受对象,并过滤字段.

2.重命名属性

出于隐藏后台数据,保证安全的角度考虑,可以使用attribute 配置面向公众的字段.比如

resource_fields={
 'name' = fields.String(attribute='true_name')
}

# 任何可调用的函数,比如 lambda 也可以指定 attribute

resource_fields={
 'name' = fields.String(attribute=lambda x:x.true_name)
}
# 也可以指定嵌套属性
resource_fields = {
 'name': fields.String(attribute='people_list.0.person_dictionary.name'),
 'address': fields.String,
}

3.默认值

可以指定默认值,这样就不会返回None

resource_fields = {
 'name':fields.String(default='any')
}

4.自定义字段和值

如果需要自定义格式,可以继承自fields.Raw 类并重载format 方法.比如:

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"

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

5.fields.Url

fields.Url:为请求的资源综合一个URL .类似如下

from flask_restful import Resource,fields,marshal_with,Api
from flask import Flask
from datetime import datetime
# pytz是一个时区的datetime object
import pytz

app = Flask(__name__)

# 1.Api绑定app
api = Api(app)

class Register:
 def __init__(self,name):
     self.name = name
     self.date = datetime.now(tz=pytz.timezone('Asia/Shanghai'))

# 2.重载Get/Post方法
class Todo(Resource):

 resource_fields = {
     'name':fields.String,
     'date': fields.DateTime(dt_format='rfc822'),
     # 返回一个url 相对路径
     'uri': fields.Url('todo'),
     # 返回一个url 绝对路径
     'uri_absolute': fields.Url('todo',absolute=True),
     # 返回一个url http绝对路径
     'https_uri':fields.Url('todo',absolute=True,scheme='https')
 }

 @marshal_with(resource_fields)
 def get(self,name):
     print(name)
     return Register(name)

# 3.注册路由/资源
api.add_resource(Todo,'/api/todo/<name>/',endpoint='todo')

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

访问:

❯ curl http://127.0.0.1:5000/api/todo/Jack/
{
 "name": "Jack",
 "date": "Thu, 12 Dec 2019 11:14:05 -0000",
 "uri": "/api/todo/Jack/",
 "uri_absolute": "http://127.0.0.1:5000/api/todo/Jack/",
 "https_uri": "https://127.0.0.1:5000/api/todo/Jack/"
}

6.复杂结构

引入2个方法:

  • flask_restful.marshal(data,fields,envelope=None):把dict\list 数据类型转换为OrderedDict 数据类型,用于过滤数据.
# data:实际的对象
# fields: 需要过滤的数据类型
# envelope: 类似别名

# 原始数据,需要过滤的数据
>>> data = {'a':100,'b':'foo'}
# 过滤条件
>>> mfields = {'a':fields.Raw}
# 过滤
>>> marshal(data,mfields)
OrderedDict([('a', 100)])

# 过滤后的数据,可以打包成 json数据
>>> import json
>>> json.dumps(marshal(data,mfields))
'{"a": 100}'
  • flask_restful.marshal_with(fields, envelope=None) :一个装饰器,可以支持自定义函数,返回的效果和上一个相同.
>>> from flask_restful import marshal_with
>>> mfields = {'a':fields.String}
>>> @marshal_with(mfields)
... def get():
...     return {'a':'foo','b':100}
...     
... 
>>> 
>>> get()
OrderedDict([('a', 'foo')])
>>> json.dumps(get())
'{"a": "foo"}'

处理一个复杂的结构:

# 过滤器,过滤条件,输出格式.
>>> resource_fields = {
...     'name':fields.String,
...     'address': {
...         'line1':fields.String,
...         'line2':fields.Integer,
...         'line3':fields.Raw
...     }
... }

# 原始数据
>>> data = {
...     'name':'Jack',
...     'line1':'This is a test',
...     'line2':123,
...     'line3':'hello'
... }

# 组装成 OrderDict对象
>>> marshal(data,resource_fields)
OrderedDict([('name', 'Jack'), ('address', OrderedDict([('line1', 'This is a test'), ('line2', 123), ('line3', 'hello')]))])

# 生成Json数据
>>> json.dumps(marshal(date,resource_fields))
'{"name": "jack", "address": {"line1": null, "line2": 0, "line3": null}}'

也可以对列表进行处理:

# 需要处理的数据
>>> data = {
...     'book':'Python',
...     'detail':['price','description']
... }

# 过滤器
>>> resource_fields = {
...     'name': fields.String,
...     'detail':fields.List(fields.String)
... }

# 转化
>>> marshal(data,resource_fields)
OrderedDict([('name', None), ('detail', ['price', 'description'])])

# Json数据
>>> json.dumps(marshal(data,resource_fields))
'{"name": null, "detail": ["price", "description"]}'
  • fiels.List() 只能过滤一种数据类型

处理嵌套数据fields.Nested

>>> data = {
...     'book':'Python',
...     'detail':{
...         'price':12,
...         'description':'This is a Python book.'
...     }
... }
>>> 
>>> resource_fields = {
...     'name': fields.String,
...     'detail':fields.List(fields.Nested({
...         'price':fields.Integer,
...         'description':fields.String
...     }))
... }
>>> 
>>> marshal(data,resource_fields)
OrderedDict([('name', None), ('detail', [OrderedDict([('price', 12), ('description', 'This is a Python book.')])])])
>>> json.dumps(marshal(data,resource_fields))
'{"name": null, "detail": [{"price": 12, "description": "This is a Python book."}]}'

7.实际使用

通过定义一个API ,集合数据库,返回给前端数据库中的数据.

.
├── app.py
├── config.py
├── exts.py
├── manage.py
├── migrations
│   ├── alembic.ini
│   ├── env.py
│   ├── README
│   ├── script.py.mako
│   └── versions
│       ├── 07d0ca1a599d_.py
├── static
└── templates

config.py

DEBUG=True
TEMPLATES_AUTO_RELOAD=True
DB_URI = 'mysql+pymysql://root:2008.Cn123@192.168.0.101:3306/flask_restful_demo' # 确保数据库存在
# 指定数据库连接
SQLALCHEMY_DATABASE_URI = DB_URI
SQLALCHEMY_TRACK_MODIFICATIONS = False

exts.py

from flask_sqlalchemy import SQLAlchemy

db = SQLAlchemy()

sql_models.py

from exts import db

class Users(db.Model):
 __tablename__ = 'users'

 id = db.Column(db.Integer,primary_key=True,nullable=False,autoincrement=True)
 name = db.Column(db.String(40),nullable=False)
 email = db.Column(db.String(40),nullable=False)
 password = db.Column(db.String(40),nullable=False)

 # 创建双向引用
 articles = db.relationship('Articles', back_populates='users')

article_tag = db.Table('article_tag',
                    db.Column('a_id',db.Integer,db.ForeignKey('articles.id'),primary_key=True),
                    db.Column('t_id',db.Integer,db.ForeignKey('tags.id'),primary_key=True)
                    )


class Articles(db.Model):
 __tablename__ = 'articles'

 id = db.Column(db.Integer, primary_key=True, nullable=False, autoincrement=True)
 name = db.Column(db.String(40), nullable=False)
 contents = db.Column(db.Text,nullable=False)

 # 与表users是多对一的关系
 u_id = db.Column(db.Integer,db.ForeignKey('users.id'))

 # 创建双向引用
 users= db.relationship('Users', back_populates='articles')
 tags = db.relationship('Tags',secondary=article_tag ,back_populates='articles')


class Tags(db.Model):
 __tablename__ = 'tags'

 id = db.Column(db.Integer, primary_key=True, nullable=False, autoincrement=True)
 name = db.Column(db.String(40), nullable=False)

 # 创建双向引用
 articles = db.relationship('Articles',secondary=article_tag, back_populates='tags')

初始化数据库,并且生成迁移脚本

mysql>  create database flask_restful_demo;

初始化,并生成迁移脚本.

from flask_script import Manager
from app import app
from flask_migrate import Migrate,MigrateCommand
from exts import db
# 导入ORM
import sql_models

# 绑定app
manage = Manager(app)

# 导入 flask-migrate
# 导入的Migrate类 可以绑定app ,db到 migrate 中
# 导入的MigrateCommand 类可以使用 Alembic 中所有的命令
Migrate(app,db)

manage.add_command('db', MigrateCommand)

if __name__ == '__main__':
 manage.run()
> python manage.py db init
> python manage.py db migrate
> python manage.py db upgrade

app.py

from flask import Flask, views
import config
from exts import db
from flask_restful import Api, Resource, marshal_with, fields
from sql_models import Users, Articles, Tags

app = Flask(__name__)
app.config.from_object(config)

# db绑定app
db.init_app(app)
# Api绑定app
api = Api(app)


@app.route('/')
def index():
    return '这是主页'


@app.route('/register/')
def register():
    '''注册数据'''
    user = Users(name='Jack', email='Jack@kning.com', password='123')
    article = Articles(name='python', contents='First paper', u_id=1)
    tag1 = Tags(name="Python")
    tag2 = Tags(name='Computer')
    article.users = user
    article.tags.append(tag1)
    article.tags.append(tag2)
    db.session.add(article)
    db.session.commit()
    return '注册成功'


# 重载方法GET,并且格式化数据输出
class ArticleOutput(Resource):
    resource_fields = {   
        # 隐藏原始数据的 字段
        'article_name': fields.String(attribute='name'),
        'article_contents': fields.String(attribute='contents'),
        'article_author': fields.String(fields.Nested({
            'author_name': fields.String(attribute='name'),
            'author_email': fields.String(attribute='email')
        })),
        'article_tags': fields.String(fields.Nested({
            'tag1': fields.String(attribute='name1'),
            'tag2': fields.String(attribute='name2')
        }))
    }

    # marshal_with 重载数据为 OrderDict.
    @marshal_with(resource_fields)
    def get(self, article_id):
        article_id = int(article_id)
        article = db.session.query(Articles).filter(Articles.id==article_id).first()

        # 原始数据
        detail = {
            'name': article.name,
            'contents': article.contents,
            #'article_name':'Python',
            #'article_contents':'first page',
            'article_author': {
                'name': article.users.name,
                'email': article.users.email
            },
            'article_tags': {
                'name1': article.tags[0].name,
                'name2': article.tags[-1].name
            }

        }
        return detail

api.add_resource(ArticleOutput, '/api/article/<int:article_id>', endpoint='article')


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

访问网站

❯ curl http://127.0.0.1:5000/api/article/1
{
    "article_name": "python",
    "article_contents": "First paper",
    "article_author": "{'name': 'Jack', 'email': 'Jack@kning.com'}",
    "article_tags": "{'name1': 'Python', 'name2': 'Computer'}"
}

8.蓝图中使用

在蓝图中使用时,初始化的对象不是app,而是蓝图bp

article_bp = Blueprint('article',__name__,url_prefix='/article')
api = Api(article_bp)

9.模板渲染

flask_restful是一个API 的实现,它返回的都是json 数据,如果要使用html ,需要先声明.

@api.representation('text/html')
def out_html(data, code, headers):
 resp = make_response(data)
 return resp


class HelloView(Resource):
 def get(self):
     return render_template('hello.html')

api.add_resource(HelloView, '/hello/', endpoint='hello')
文章作者: 文彦
文章链接: https://www.kningyuan.top/post/10069.html
版权声明: 本博客所有文章除特別声明外，均采用 CC BY 4.0 许可协议。转载请注明来源 文彦 !
Flask
赏
评论
  目录