使用python3和flask构建RESTful API(接口测试服务与mockserver工具)

引言

构建RESTful API貌似是开发的工作，和测试有和关系？

其实测试开发需要构建RESTful API的场景很多。比如测试Android应用，一般的接口测试只考虑了服务器端，至于客户端在网络异常或者服务端异常时如何反应，多数天朝的测试人员是没有考虑到的。客户端在对这些异常处理不够充分的时候，会出现崩溃等各种莫名其妙的问题。

为此一些走在前沿的测试人员会自己写一些RESTful API， 把服务端的域名劫持到自己的API，故意返回各种异常，看客户端的稳定性。

另外测试开发的测试工具需要和其他系统对接等场景也经常需要API。

参考资料

• How to Build RESTful APIs with Python and Flask
• 本文地址
• 本文源码地址
• 讨论qq群144081101 591302926 567351477 钉钉免费群21745728
• 本文涉及的python测试开发库 谢谢点赞！
• 本文相关书籍下载

术语

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

要求

• python3.*
• PostgreSQL

工程目录

project/├── app.py├── config.py├── migrate.py├── Model.py├── requirements.txt├── resources│   └── Hello.py│   └── Comment.py│   └── Category.py└── run.py

requirements.txt的内容如下：

flaskflask_restfulflask_scriptflask_migratemarshmallowflask_sqlalchemyflask_marshmallowmarshmallow-sqlalchemypsycopg2-binarypython-daemon

• 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。

安装依赖

# pip3 install -r requirements.txt

安装配置PostgreSQL

这里以 Ubuntu 16.04为例：

# 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

配置

# -*- coding: utf-8 -*-# Author:    xurongzhong#126.com wechat:pythontesting qq:37391319# CreateDate: 2018-1-10from flask import Blueprintfrom flask_restful import Apifrom resources.Hello import Hellofrom resources.Category import CategoryResourcefrom resources.Comment import CommentResourceapi_bp = Blueprint('api', __name__)api = Api(api_bp)# Routesapi.add_resource(Hello, '/Hello')api.add_resource(CategoryResource, '/Category')api.add_resource(CommentResource, '/Comment')

快速入门

app.py

from flask import Blueprintfrom flask_restful import Apifrom resources.Hello import Helloapi_bp = Blueprint('api', __name__)api = Api(api_bp)# Routeapi.add_resource(Hello, '/Hello')

resource/Hello.py

#!/usr/bin/python# -*- coding: utf-8 -*-# Author:    xurongzhong#126.com wechat:pythontesting qq:37391319# CreateDate: 2018-1-10from flask_restful import Resourceclass Hello(Resource):    def get(self):        return {"message": "Hello, World!"}    def post(self):        return {"message": "Hello, World!"}

run.py

from flask import Flaskdef 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 appif __name__ == "__main__":    app = create_app("config")    app.run(debug=True)

启动服务

$ 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

{    "hello": "world"}

接入数据库

from flask import Flaskfrom marshmallow import Schema, fields, pre_load, validatefrom flask_marshmallow import Marshmallowfrom flask_sqlalchemy import SQLAlchemyma = 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_idclass 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 = nameclass 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

from flask_script import Managerfrom flask_migrate import Migrate, MigrateCommandfrom Model import dbfrom run import create_appapp = create_app('config')migrate = Migrate(app, db)manager = Manager(app)manager.add_command('db', MigrateCommand)if __name__ == '__main__':    manager.run()

数据迁移

$ python3 migrate.py db init$ python3 migrate.py db migrate$ python migrate.py db upgrade

修改Category.py 和Comment.py， 完整代码

测试

可以使用curl，比如：

curl http://127.0.0.1:5000/api/Category --data '{"name":"test5","id":5}' -H "Content-Type: application/json"

也可以在chrome中使用postman：

image

image

image