Flask RESTful API 简单的设计一个 GET 请求接口

数据库：PostgreSQL

框架：Flask

语言：Python 3.6

前提

之前我有每天定时爬取 bing 壁纸，写入 postgresql 数据库的，如下：

ER 图

需要的 Python 环境

flask-sqlalchemy
flask-migrate
flask-script
flask-restful
flask
psycopg2

创建配置文件 Config.py

里面设置数据库的引擎以及其它的一些需要配置的程序参数

#!/usr/bin/env python
# -*- coding: utf-8 -*-
# @Time    : 18/6/25 下午 5:15
# @Author  : olei
# @File    : Config.py
# @Software: PyCharm
# @license : Copyright(C), iicats.com
# @Contact : i@olei.me

DB_USER = "postgres"
DB_PASSWORD = "postgres"
HOST = "127.0.0.1"
DB_NAME = "bing"


DEBUG = True

SQLALCHEMY_TRACK_MODIFICATIONS = False
SQLALCHEMY_DATABASE_URI = 'postgresql://' + DB_USER + ':' + DB_PASSWORD + "@" + HOST + '/' + DB_NAME

这里设置的是 postgresql 的配置项

数据模型 Model.py

根据数据库，来设置数据模型

#!/usr/bin/env python
# -*- coding: utf-8 -*-
# @Time    : 18/6/25 下午 5:22
# @Author  : olei
# @File    : Model.py
# @Software: PyCharm
# @license : Copyright(C), iicats.com
# @Contact : i@olei.me

from flask_sqlalchemy import SQLAlchemy


db = SQLAlchemy()

class Bing(db.Model):
    __tablename__ = "psql_bing"
    id = db.Column(db.Integer,unique=True,primary_key=True)
    dates = db.Column(db.Date,nullable=False)
    bing_url = db.Column(db.String(10000),nullable=False)
    qiniu_url = db.Column(db.String(10000),nullable=False)
    image_name = db.Column(db.String(10000),nullable=False)

忽略我给的 String 的大小...，这个与 flask 中设计表结构的写法是一致的，用到 flask 的 flask-sqlalchemy

数据库迁移文件 migrate.py

使用 Flask-Migrate 和 Flask-Script 来实现数据迁移

#!/usr/bin/env python
# -*- coding: utf-8 -*-
# @Time    : 18/6/25 下午 5:25
# @Author  : olei
# @File    : migrate.py
# @Software: PyCharm
# @license : Copyright(C), iicats.com
# @Contact : i@olei.me

from flask import Flask
from flask_migrate import Migrate,MigrateCommand
from flask_script import Manager
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()

这里的 app.config.from_object 是读取配置文件，里面写上配置文件的没有拓展名的文件名，就是上面的配置文件 Config.py 的名字

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

$ python migrate.py db init

$ python migrate.py db migrate

$ python migrate.py db upgrade

结束之后会生成一个文件夹，里面也会生成一些文件，如下：

migrations
├── README
├── alembic.ini
├── env.py
├── script.py.mako
└── versions
    └── 79d73a8da1cf_.py

Flask-RESTful 接口实现 app.py

先给出代码

#!/usr/bin/env python
# -*- coding: utf-8 -*-
# @Time    : 18/6/26 上午 11:21
# @Author  : olei
# @File    : app.py
# @Software: PyCharm
# @license : Copyright(C), iicats.com
# @Contact : i@olei.me

from flask import Flask, jsonify
from Model import db, Bing
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 = {
    'BingAlreadyExistsError': {
        'message': "It is 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('dates', required=True)
parser.add_argument('bing_url', required=True)
parser.add_argument('qiniu_url', required=True)
# parser.add_argument('user_nickname')
parser.add_argument('image_name', required=True)

resource_full_fields = {
    'id': fields.Integer,
    'dates': fields.String,
    'bing_url': fields.String,
    'qiniu_url': fields.String,
    'image_name': 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 Bing_all(Resource):
    def get(self):
        # dates = Bing.query.filter_by()
        return Common.returnTrueJson(Common, marshal(Bing.query.all(), resource_full_fields))


class Bing_url(Resource):
    def get(self, dates):
        dates = Bing.query.filter_by(dates=dates).first()
        if (dates is None):
            abort(410, msg="找不到数据！", data=None, status=0)
        else:
            return Common.returnTrueJson(Common, marshal(dates, resource_full_fields))


api.add_resource(Hello, '/', '/hello')
api.add_resource(Bing_all, '/bing')
api.add_resource(Bing_url, '/bing/<string:dates>')

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

分析

错误处理代码

errors = {
    'BingAlreadyExistsError': {
        'message': "It is 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 来调用的

 if (dates is None):
     abort(410, msg="找不到数据！", data=None, status=0)

定义一个 Common，来统一响应数据格式
处理请求

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

parser = reqparse.RequestParser()
parser.add_argument('dates', required=True)
parser.add_argument('bing_url', required=True)
parser.add_argument('qiniu_url', required=True)
# parser.add_argument('user_nickname')
parser.add_argument('image_name', required=True)