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

引言

构建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

要求

  • 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
「点点赞赏,手留余香」

    还没有人赞赏,快来当第一个赞赏的人吧!
0 条回复 A 作者 M 管理员
    所有的伟大,都源于一个勇敢的开始!
欢迎您,新朋友,感谢参与互动!欢迎您 {{author}},您在本站有{{commentsCount}}条评论
个人博客 论坛专区 主机两折 爆款主机
关注我们
  • 官方客服:289553844一键联系
  • 扫一扫加公众号
    扫一扫加小程序