引言
构建RESTful API貌似是开发的工作，和测试有和关系？
其实测试开发需要构建RESTful API的场景很多。比如测试Android应用，一般的接口测试只考虑了服务器端，至于客户端在网络异常或者服务端异常时如何反应，多数天朝的测试人员是没有考虑到的。客户端在对这些异常处理不够充分的时候，会出现崩溃等各种莫名其妙的问题。
为此一些走在前沿的测试人员会自己写一些RESTful API， 把服务端的域名劫持到自己的API，故意返回各种异常，看客户端的稳定性。
另外测试开发的测试工具需要和其他系统对接等场景也经常需要API。
参考资料
术语
REST： REpresentational State Transfer
目标
GET - /api/Category - Retrieve all categories
POST - /api/Category - Add a new category
PUT - /api/Category - Update a category
DELETE - /api/Category - Delete a category
GET - /api/Comment - Retrieve all the stored comments
POST - /api/Comment - Add new comment
要求
工程目录
#!python
project/
├── app.py
├── config.py
├── migrate.py
├── Model.py
├── requirements.txt
├── resources
│ └── Hello.py
│ └── Comment.py
│ └── Category.py
└── run.py
requirements.txt的内容如下：
#!python
flask
flask_restful
flask_script
flask_migrate
marshmallow
flask_sqlalchemy
flask_marshmallow
marshmallow-sqlalchemy
psycopg2
flask - Python的微框架
flask_restful - 这是Flask的扩展，可快速构建REST API。
flask_script - 提供了在Flask中编写外部脚本的支持。
flask_migrate - 使用Alembic的Flask应用进行SQLAlchemy数据库迁移。
marshmallow - ORM/ODM/框架无关的库，用于复杂数据类型（如对象）和Python数据类型转换。
flask_sqlalchemy - Flask扩展，增加了对SQLAlchemy的支持。
flask_marshmallow - 这是Flask和marshmallow的中间层。
marshmallow-sqlalchemy - 这是sqlalchemy和marshmallow的中间层。
psycopg - Python的PostgreSQL API。
安装依赖
#!python
# pip3 install -r requirements.txt
安装配置PostgreSQL
这里以 Ubuntu 16.04为例：
#!python
# sudo apt-get update && sudo apt-get upgrade
# apt-get install postgresql postgresql-contrib
# su - postgres
$ createdb api
$ createuser andrew --pwprompt #创建用户
$ psql -d api -c "ALTER USER andrew WITH PASSWORD 'api';"
参考资料：
How to Install PostgreSQL on Ubuntu 16.04
How To Install and Use PostgreSQL on Ubuntu 14.04
配置
#!python
# -*- coding: utf-8 -*-
# Author: xurongzhong#126.com wechat:pythontesting qq:37391319
# CreateDate: 2018-1-10
from flask import Blueprint
from flask_restful import Api
from resources.Hello import Hello
from resources.Category import CategoryResource
from resources.Comment import CommentResource
api_bp = Blueprint('api', __name__)
api = Api(api_bp)
# Routes
api.add_resource(Hello, '/Hello')
api.add_resource(CategoryResource, '/Category')
api.add_resource(CommentResource, '/Comment')
快速入门
app.py
#!python
from flask import Blueprint
from flask_restful import Api
from resources.Hello import Hello
api_bp = Blueprint('api', __name__)
api = Api(api_bp)
# Route
api.add_resource(Hello, '/Hello')
resource/Hello.py
#!python
#!/usr/bin/python
# -*- coding: utf-8 -*-
# Author: xurongzhong#126.com wechat:pythontesting qq:37391319
# CreateDate: 2018-1-10
from flask_restful import Resource
class Hello(Resource):
def get(self):
return {"message": "Hello, World!"}
def post(self):
return {"message": "Hello, World!"}
run.py
#!python
from flask import Flask
def create_app(config_filename):
app = Flask(__name__)
app.config.from_object(config_filename)
from app import api_bp
app.register_blueprint(api_bp, url_prefix='/api')
return app
if __name__ == "__main__":
app = create_app("config")
app.run(debug=True)
启动服务
#!python
$ python3 run.py
* Running on http://127.0.0.1:5000/ (Press CTRL+C to quit)
* Restarting with stat
* Debugger is active!
* Debugger PIN: 136-695-873
用浏览器访问： http://127.0.0.1:5000/api/Hello
#!python
{
"hello": "world"
}
接入数据库
#!python
from flask import Flask
from marshmallow import Schema, fields, pre_load, validate
from flask_marshmallow import Marshmallow
from flask_sqlalchemy import SQLAlchemy
ma = Marshmallow()
db = SQLAlchemy()
class Comment(db.Model):
__tablename__ = 'comments'
id = db.Column(db.Integer, primary_key=True)
comment = db.Column(db.String(250), nullable=False)
creation_date = db.Column(db.TIMESTAMP, server_default=db.func.current_timestamp(), nullable=False)
category_id = db.Column(db.Integer, db.ForeignKey('categories.id', ondelete='CASCADE'), nullable=False)
category = db.relationship('Category', backref=db.backref('comments', lazy='dynamic' ))
def __init__(self, comment, category_id):
self.comment = comment
self.category_id = category_id
class Category(db.Model):
__tablename__ = 'categories'
id = db.Column(db.Integer, primary_key=True)
name = db.Column(db.String(150), unique=True, nullable=False)
def __init__(self, name):
self.name = name
class CategorySchema(ma.Schema):
id = fields.Integer()
name = fields.String(required=True)
class CommentSchema(ma.Schema):
id = fields.Integer(dump_only=True)
category_id = fields.Integer(required=True)
comment = fields.String(required=True, validate=validate.Length(1))
creation_date = fields.DateTime()
migrate.py
#!python
from flask_script import Manager
from flask_migrate import Migrate, MigrateCommand
from Model import db
from run import create_app
app = create_app('config')
migrate = Migrate(app, db)
manager = Manager(app)
manager.add_command('db', MigrateCommand)
if __name__ == '__main__':
manager.run()
数据迁移
#!python
$ python3 migrate.py db init
$ python3 migrate.py db migrate
$ python migrate.py db upgrade
修改Category.py 和Comment.py， 完整代码
测试
可以使用curl，比如：
#!python
curl http://127.0.0.1:5000/api/Category --data '{"name":"test5","id":5}' -H "Content-Type: application/json"
也可以在chrome中使用postman：
![]()
how_to_build_restful_apis_using_flask_creating_categories.png
![]()
how_to_build_restful_apis_using_flask_listing_categories.png
![]()
how_to_build_restful_apis_using_flask_creating_comment.png
![]()
how_to_build_restful_apis_in_flask_listing_comments.png
