[雪峰磁针石博客]使用python3和flask构建RESTful API(接口测试服务)

简介: 引言 构建RESTful API貌似是开发的工作,和测试有和关系? 其实测试开发需要构建RESTful API的场景很多。比如测试Android应用,一般的接口测试只考虑了服务器端,至于客户端在网络异常或者服务端异常时如何反应,多数天朝的测试人员是没有考虑到的。

引言

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

工程目录

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

requirements.txt的内容如下:

#!python

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。

安装依赖

#!python

# pip3 install -r requirements.txt

安装配置PostgreSQL

这里以 Ubuntu 16.04为例:

#!python

# 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

配置

#!python

# -*- 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

#!python

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

#!python

#!/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

#!python

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)

启动服务

#!python

$ 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

#!python

{
    "hello": "world"
}

接入数据库

#!python

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

#!python

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()

数据迁移

#!python

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

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

测试

可以使用curl,比如:

#!python

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

相关实践学习
使用PolarDB和ECS搭建门户网站
本场景主要介绍如何基于PolarDB和ECS实现搭建门户网站。
阿里云数据库产品家族及特性
阿里云智能数据库产品团队一直致力于不断健全产品体系,提升产品性能,打磨产品功能,从而帮助客户实现更加极致的弹性能力、具备更强的扩展能力、并利用云设施进一步降低企业成本。以云原生+分布式为核心技术抓手,打造以自研的在线事务型(OLTP)数据库Polar DB和在线分析型(OLAP)数据库Analytic DB为代表的新一代企业级云原生数据库产品体系, 结合NoSQL数据库、数据库生态工具、云原生智能化数据库管控平台,为阿里巴巴经济体以及各个行业的企业客户和开发者提供从公共云到混合云再到私有云的完整解决方案,提供基于云基础设施进行数据从处理、到存储、再到计算与分析的一体化解决方案。本节课带你了解阿里云数据库产品家族及特性。
相关文章
|
9月前
|
JSON 算法 API
Python采集淘宝商品评论API接口及JSON数据返回全程指南
Python采集淘宝商品评论API接口及JSON数据返回全程指南
|
9月前
|
JSON API 数据安全/隐私保护
Python采集淘宝拍立淘按图搜索API接口及JSON数据返回全流程指南
通过以上流程,可实现淘宝拍立淘按图搜索的完整调用链路,并获取结构化的JSON商品数据,支撑电商比价、智能推荐等业务场景。
|
10月前
|
JSON API 数据安全/隐私保护
Python采集淘宝评论API接口及JSON数据返回全流程指南
Python采集淘宝评论API接口及JSON数据返回全流程指南
|
10月前
|
缓存 监控 供应链
唯品会自定义 API 自定义操作深度分析及 Python 实现
唯品会开放平台提供丰富API,支持商品查询、订单管理、促销活动等电商全流程操作。基于OAuth 2.0认证机制,具备安全稳定的特点。通过组合调用基础接口,可实现数据聚合、流程自动化、监控预警及跨平台集成,广泛应用于供应链管理、数据分析和智能采购等领域。结合Python实现方案,可高效完成商品搜索、订单分析、库存监控等功能,提升电商运营效率。
|
10月前
|
缓存 监控 供应链
京东自定义 API 操作深度分析及 Python 实现
京东开放平台提供丰富API接口,支持商品、订单、库存等电商全链路场景。通过自定义API组合调用,可实现店铺管理、数据分析、竞品监控等功能,提升运营效率。本文详解其架构、Python实现与应用策略。
缓存 监控 供应链
352 0
缓存 监控 数据挖掘
229 0
JSON 监控 API
381 0
|
11月前
|
JSON 测试技术 API
深度分析爱回收API接口,用Python脚本实现
爱回收(Aihuishou)是国内领先的电子产品回收与以旧换新平台,提供设备估价、订单管理、物流跟踪、结算等全链路API服务,支持企业客户构建回收业务系统。需通过企业合作申请接口权限,本文详解其API体系、认证机制及Python调用方案。
|
11月前
|
供应链 JavaScript API
深度分析电子元件API接口,用Python脚本实现
电子元件API为电子制造、研发及供应链提供元件查询、库存、价格、供应商及技术文档等核心功能,支持采购决策与研发选型。主流平台包括国际的Digikey、Mouser及国内的立创商城、华强电子网,接口设计各有差异但功能逻辑一致。

推荐镜像

更多