使用python3和flask構建RESTful API(介面測試服務與mockserver工具)
引言
構建RESTful API貌似是開發的工作，和測試有和關係？
其實測試開發需要構建RESTful API的場景很多。比如測試Android應用，一般的介面測試只考慮了伺服器端，至於客戶端在網路異常或者服務端異常時如何反應，多數天朝的測試人員是沒有考慮到的。客戶端在對這些異常處理不夠充分的時候，會出現崩潰等各種莫名其妙的問題。
為此一些走在前沿的測試人員會自己寫一些RESTful API， 把服務端的域名劫持到自己的API，故意返回各種異常，看客戶端的穩定性。
另外測試開發的測試工具需要和其他系統對接等場景也經常需要API。
參考資料
- ofollow,noindex">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的內容如下：
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';"
參考資料：
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-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 -*-
# 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
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
用瀏覽器訪問： http://127.0.0.1:5000/api/Hello
{
"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：
image
image
image