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


声明:本文转载自https://my.oschina.net/u/1433482/blog/1605135,转载目的在于传递更多信息,仅供学习交流之用。如有侵权行为,请联系我,我会及时删除。

引言

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

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

本文发表于2018年01月10日 16:32
(c)注:本文转载自https://my.oschina.net/u/1433482/blog/1605135,转载目的在于传递更多信息,并不代表本网赞同其观点和对其真实性负责。如有侵权行为,请联系我们,我们会及时删除.

阅读 2579 讨论 0 喜欢 0

抢先体验

扫码体验
趣味小程序
文字表情生成器

闪念胶囊

你要过得好哇,这样我才能恨你啊,你要是过得不好,我都不知道该恨你还是拥抱你啊。

直抵黄龙府,与诸君痛饮尔。

那时陪伴我的人啊,你们如今在何方。

不出意外的话,我们再也不会见了,祝你前程似锦。

这世界真好,吃野东西也要留出这条命来看看

快捷链接
网站地图
提交友链
Copyright © 2016 - 2021 Cion.
All Rights Reserved.
京ICP备2021004668号-1