AireadFan's

   Fragments of memories

Airead-manager-outline-step5

| Comments

添加增删改查 restful 接口

添加 rest 接口

blueprints/rest 中创建文件 persons.py, 注意这里有s

(persons.py) download
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
# coding=utf-8
from aireadManager.utils.errors import Code
import os
from flask.blueprints import Blueprint
from flask import request, abort
from flask.ext.restful import Api, reqparse, fields, marshal_with
from aireadManager.utils.restful import Resource
from aireadManager.model.person import PersonModel
from aireadManager.model import db

__author__ = 'airead'

path = os.path.splitext(os.path.basename(__file__))[0]
blueprint = Blueprint(path, __name__, url_prefix='/' + path)
api = Api(blueprint)  # 获取 person 蓝图 rest api 实例


person_fields = {  # used by marshal_with, see flask-restful
    'id': fields.Integer,
    'name': fields.String,
}


def get_parser(required=False):  # 用来解析 request.values, 参考 flask-restful reqparse
    parser = reqparse.RequestParser()
    parser.add_argument('name', type=unicode, required=required)

    return parser


class Persons(Resource):  # Persons rest class, 注意要加s
    @marshal_with(person_fields)
    def get(self):  # get method
        persons = PersonModel.query.all()
        print persons
        return persons

    def post(self):  # post method
        print request.form
        args = get_parser(required=True).parse_args()  # 获取 Person 信息

        person = PersonModel(**args)  # see sqlalchemy
        db.session.add(person)
        db.session.commit()

        return {
            'code': Code.SUCCESS,
            'id': person.id,
            'uri': api.url_for(Person, pid=person.id)
        }


class Person(Resource):  # Person rest class, 注意没有s
    @marshal_with(person_fields)
    def get(self, pid):  # 这里的 pid 与 add_resource 中的 <int:pid> 对应
        return db.session.query(PersonModel).filter_by(id=pid).first()

    # 只是为了充许使用 post 方法,因为存在浏览器不能使用 put, delete 方法,所以
    # delete 与 put 方法通过 post url 中的 ?at=delete, ?at=put 来识别
    def post(self, pid):
        abort(405)

    def put(self, pid):
        args = get_parser().parse_args()

        set_data = {key: val for key, val in args.iteritems() if val is not None}  # 获取要修改的值
        print 'set_data', set_data

        db.session.query(PersonModel).filter_by(id=pid).update(set_data)
        db.session.commit()
        return {'code': Code.SUCCESS}  # 通用返回成功格式

    def delete(self, pid):
        db.session.query(PersonModel).filter_by(id=pid).delete()
        db.session.commit()

        return {'code': Code.SUCCESS}


api.add_resource(Persons, '/', endpoint='.persons')  # 添加 rest 路由, 通过 /persons/ 访问
api.add_resource(Person, '/<int:pid>', endpoint='.person')  # 通过 /persons/1 访问

测试 rest 接口

为了验证 persons rest 接口的可用性,下面使用 nose 结何 flask-testing 进行测试

test 中创建文件 test_persons.py

(test_persons.py) download
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
# coding=utf-8
from flask.ext.testing import TestCase
from nose.tools import assert_equal
from datetime import datetime

from aireadManager.main import app
from aireadManager.model import db
from aireadManager.model.person import PersonModel
from aireadManager.utils.errors import Code
from aireadManager.utils.util import get_string_from_datetime
from aireadManager.utils.principal import role_set
from aireadManager.utils.permissions import Roles
from aireadManager.test.init_te_st_db import USER1, USER2, init_db, TestDBConfig


__author__ = 'airead'

SuccessRet = {
    'code': Code.SUCCESS
}

person1 = {'name': 'airead'}  # 模拟数据库数据
person2 = {'name': 'fan'}


class Test_persons(TestCase):  # see python unittest
    def create_app(self):
        app.config.from_object(TestDBConfig)
        return app

    def setUp(self):
        db.create_all()
        for person in [person1, person2]:  # 每次测试开始添加两条记录到 persons 表
            _p = PersonModel(**person)
            db.session.add(_p)
        db.session.commit()

    def tearDown(self):
        db.session.remove()
        db.drop_all()  # 每次测试结束清空数据库

    def test_persons_get(self):  # test /persons, 获取 persons
        rv = self.client.get('/persons/')

        p1, p2 = rv.json
        assert_equal(p1['id'], 1)
        assert_equal(p1['name'], person1['name'])
        assert_equal(p2['id'], 2)
        assert_equal(p2['name'], person2['name'])

    def test_persons_post(self):  # /persons/ 测试添加 person,
        data = {
            'name': 'man',
        }
        rv = self.client.post('persons/', data=data)
        print rv.data
        assert_equal(rv.json['uri'], '/persons/3')

        ps = db.session.query(PersonModel).all()
        assert_equal(len(ps), 3)

    def test_person_put(self):  # /persons/1?at=put 测试修改 person
        data = {
            'name': 'Hao'
        }
        rv = self.client.post('persons/1?at=put', data=data)
        assert_equal(rv.json, SuccessRet)

        p = db.session.query(PersonModel).filter_by(id=1).one()
        assert_equal(p.name, 'Hao')

    def test_person_delete(self):  # 测试通过 id 删除 person
        rv = self.client.post('persons/1?at=delete')
        assert_equal(rv.json, SuccessRet)

        ps = db.session.query(PersonModel).all()
        assert_equal(len(ps), 1)
        p = ps[0]
        assert_equal(p.name, person2['name'])

    def test_person_get(self):  # 测试获取单个 person
        rv = self.client.get('/persons/1')
        p = rv.json
        assert_equal(p['name'], person1['name'])

运行 nosetests test/test_persons.py -v, 应该全部测试OK

添加增删改查 restful 接口完成

git checkout step5 查看源码

airead-manager

« airead-manager-outline-step4 airead-manager-outline-step6 »

Comments