使用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的內容如下:
flask flask_restful flask_script flask_migrate marshmallow flask_sqlalchemy flask_marshmallow marshmallow-sqlalchemy psycopg2-binary python-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';"
參考資料:
配置
# -*- coding: utf-8 -*-
# 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
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
#!/usr/bin/python
# -*- coding: utf-8 -*-
# qq群:548377875
# 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
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)
啟動服務
$ 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
{
"hello": "world"
}
接入資料庫
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
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()
資料遷移
$ 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: