使用Flask-RESTful快速创建RESTful API接口

Flask-RESTful是一个用于快速创建RESTful API接口的Flask扩展。使用Flask-RESTful可以很快速方便地创建一个RESTful风格的接口应用程序。

一、基本使用

先贴个代码:

from flask import Flask
from flask_restful import Resource, Api

app = Flask(__name__)

api = Api(app)

class Hello(Resource):
    def get(self):
        return 'Hello Flask!'

api.add_resource(Hello, '/')

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

上面就是Flask-RESTful一个最基本的使用示例,一共三个要点:

一是用当前程序初始化一个Flask-RESTful的API实例:

api = Api(app)

二是定义资源类:

class Hello(Resource):

def get(self):

return ‘Hello Flask!’

资源类中是RESTful接口的具体实现。

三是为RESTful接口定义接口(Endpoint)

api.add_resource(Hello, ‘/’)

运行上面程序,使用http://127.0.0.1:5000/可以打印出”Hello Flask!”

 

下面来看看比较有实际应用意义的例子。

二、一个数据库操作实例

本实例使用Flask-RESTful来实现一个用户表的增删改查操作。本示例程序一共有4个文件:

Config.py:程序配置文件

Model.py:用户表数据模型

Db.py:数据迁移配置

App.py:接口实现

1、Flask环境搭建

首先要完成Flask环境搭建,可参考《Flask环境搭建及基础项目构建》

安装必要的扩展:

pip install flask-mysqldb

pip install flask-sqlalchemy

pip install flask-migrate

pip install flask-script

pip install flask-restful

2、配置文件config.py

DB_USER = 'root'
DB_PASSWORD = ''
DB_HOST = 'localhost'
DB_DB = 'test'

DEBUG = True

SQLALCHEMY_TRACK_MODIFICATIONS = False
SQLALCHEMY_DATABASE_URI = 'mysql://' + DB_USER + ':' + DB_PASSWORD + '@' + DB_HOST + '/' + DB_DB

3、用户数据模型model.py

from flask_sqlalchemy import SQLAlchemy

db = SQLAlchemy()


class User(db.Model):
    user_id = db.Column(db.Integer, primary_key=True)
    user_name = db.Column(db.String(60), nullable=False)
    user_password = db.Column(db.String(30), nullable=False)
    user_nickname = db.Column(db.String(50))
    user_email = db.Column(db.String(30), nullable=False)

4、数据迁移配置db.py

使用Flask-Migrate和Flask-Script来实现数据迁移,具体使用方法可参考《Flask数据库迁移:Flask-Migrate和Flask-Script的使用》

from flask import Flask
from flask_script import Manager
from flask_migrate import Migrate, MigrateCommand
from model import db

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

migrate = Migrate(app, db)
manager = Manager(app)
manager.add_command('db', MigrateCommand)

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

定义好数据迁移配置后,执行下面命令完成数据迁移:

python db.py db init

python db.py db migrate

python db.py db upgrade

5、Flask-RESTful接口实现app.py

这里先给出示例代码:

from flask import Flask, jsonify
from model import db, User
from flask_restful import Resource, Api, reqparse, fields, marshal_with, abort, marshal

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

db.init_app(app)

errors = {
    'UserAlreadyExistsError': {
        'message': "A user with that username already exists.",
        'status': 409,
    },
    'ResourceDoesNotExist': {
        'message': "A resource with that ID no longer exists.",
        'status': 410,
        'extra': "Any extra information you want.",
    },
}

api = Api(app, catch_all_404s=True, errors=errors)

parser = reqparse.RequestParser()
parser.add_argument('user_name', required=True)
parser.add_argument('user_password', required=True)
parser.add_argument('user_nickname')
parser.add_argument('user_email', required=True)

resource_full_fields = {
    'user_id': fields.Integer,
    'user_name': fields.String,
    'user_email': fields.String,
    'user_nickname': fields.String
}

class Common:
    def returnTrueJson(self, data, msg="请求成功"):
        return jsonify({
            "status": 1,
            "data": data,
            "msg": msg
        })
    def returnFalseJson(self, data=None, msg="请求失败"):
        return jsonify({
            "status": 0,
            "data": data,
            "msg": msg
        })

class Hello(Resource):
    def get(self):
        return 'Hello Flask!'

class Users(Resource):
    # @marshal_with(resource_full_fields, envelope='data')
    def get(self, userId):
        user = User.query.filter_by(user_id=userId).first()
        if (user is None):
            abort(410, msg="找不到数据", data=None, status=0)
            # return Common.returnFalseJson(Common)
        else:
            return Common.returnTrueJson(Common, marshal(user, resource_full_fields))

    def delete(self, userId):
        deleteRow = User.query.filter_by(user_id=userId).delete()
        db.session.commit()
        if (deleteRow):
            return UserList.get(UserList)
        else:
            return Common.returnFalseJson(Common)

    def put(self, userId):
        args = parser.parse_args()
        user_name = args['user_name']
        user_password = args['user_password']
        user_nickname = args['user_nickname']
        user_email = args['user_email']
        try:
            user = User.query.filter_by(user_id=userId).first()
            user.user_name = user_name
            user.user_password = user_password
            user.user_nickname = user_nickname
            user.user_email = user_email
            db.session.commit()
            userId = user.user_id
            data = User.query.filter_by(user_id=userId).first()
            return Common.returnTrueJson(Common, marshal(data, resource_full_fields))
        except:
            db.session.rollback()
            db.session.flush()
            abort(409, msg="修改失败", data=None, status=0)


class UserList(Resource):
    # @marshal_with(resource_full_fields, envelope='data')
    def get(self):
        # return marshal(User.query.all(), resource_full_fields)
        return Common.returnTrueJson(Common, marshal(User.query.all(), resource_full_fields))

    def post(self):
        args = parser.parse_args()
        user_name = args['user_name']
        user_password = args['user_password']
        user_nickname = args['user_nickname']
        user_email = args['user_email']
        user = User(user_name=user_name, user_password=user_password, user_nickname=user_nickname,
                    user_email=user_email)
        try:
            db.session.add(user)
            db.session.commit()
        except:
            db.session.rollback()
            db.session.flush()
        if (user.user_id is None):
            return Common.returnFalseJson(Common, msg="添加失败")
        else:
            return Users.get(Users, user.user_id)


api.add_resource(Hello, '/', '/hello')
api.add_resource(UserList, '/users')
api.add_resource(Users, '/users/<int:userId>')

if __name__ == '__main__':
    app.run(debug=app.config['DEBUG'])

下来来分析下上面实现代码的核心部分

(1)错误处理

使用Flask-RESTful的API类,传入catch_all_404s=True(默认为False),可以由API类的handle_error()方法处理404类错误。handle_error()方法需要一个异常(Exception)作为参数。如:

 

from werkzeug.exceptions import HTTPException

api = Api(app, catch_all_404s=True)

def post(self):
    exception = HTTPException()
    exception.code = 404
    exception.data = {}
    api.handle_error(exception)

另外,也可以在API类中传入errors参数,自定义数据处理,如本例所示:

errors = {
    'UserAlreadyExistsError': {
        'message': "A user with that username already exists.",
        'status': 409,
    },
    'ResourceDoesNotExist': {
        'message': "A resource with that ID no longer exists.",
        'status': 410,
        'extra': "Any extra information you want.",
    },
}

api = Api(app, catch_all_404s=True, errors=errors)

 

调用:

abort(410, msg=”找不到数据”, data=None, status=0)

(2)示例中定义了一个Common类用来统一响应数据格式

(3)处理请求

Flask-RESTful的reqparse用于获取并转化客户端输入参数,如:

 

parser = reqparse.RequestParser()
parser.add_argument('user_name', required=True)
parser.add_argument('user_password', required=True)
parser.add_argument('user_nickname')
parser.add_argument('user_email', required=True)

上面代码定义了一个请求数据分析转化器(parser),然后指定参数的名称。

在获取参数数据时使用parse_args来转化所有的参数,并返回一个输入数据字典。

 

args = parser.parse_args()
user_name = args['user_name']
user_password = args['user_password']
user_nickname = args['user_nickname']
user_email = args['user_email']

(4)处理响应

Flask-RESTful的fields用于规范响应字段,定制响应字段键名和键值数据类型,还可以对输出响应做更多复杂的处理。

输出响应时,可以使用装饰器或函数式两种方式作处理:

装饰器方式:

@marshal_with(resource_full_fields, envelope=’data’)

函数方式:

marshal(User.query.all()

 

关于Flask-RESTful的基本介绍就到这里了,Flask-RESTful相对还是比较简单的,更多配置和用法可以看一下Flask-RESTful官方文档。本文完

 

那时那我

随遇,随缘,随安,随喜!

One thought to “使用Flask-RESTful快速创建RESTful API接口”

发表评论

电子邮件地址不会被公开。 必填项已用*标注

This site uses Akismet to reduce spam. Learn how your comment data is processed.