python

python学习之pecan框架

pecan框架的使用

Posted by YangSijie on August 15, 2018

Pecan框架

基础知识介绍：

一. 文件中需要包含一个config.py文件，该文件用于标注pecan程序的起点等配置信息：

app = {
    'root': 'webdemo.api.controllers.root.RootController',
    'modules': ['webdemo.api'],
    'debug': True,
}
  1. modules：其中包含的python包，是app.py文件所在的包，即setup_app方法所在的包
  2. root：即RootController所在的路径，/路径
  3. debug：是否开启调试，生产环境的话，将其置为False

二. pecan可以实现对象分发式的路由

当RootController继承pecan.rest.RestController时，存在URL映射关系如下表所示，当然也可以另外自己定义新的方式：

Method Description Example Method(s) / URL(s)
get_one Display one record. GET /books/1
get_all Display all records in a resource. GET /books/
get A combo of get_one and get_all. GET /books/ 或 GET /books/1
new Display a page to create a new resource. GET /books/new
edit Display a page to edit an existing resource. GET /books/1/edit
post Create a new record. POST /books/
put Update an existing record. POST /books/1?_method=put 或 PUT /books/1
get_delete Display a delete confirmation page. GET /books/1/delete
delete Delete an existing record. POST /books/1?_method=delete 或 DELETE /books/1

在RootController类中，一般会写上如下的代码：

from pecan import rest
import pecan


class RootController(rest.RestController):
    
    @pecan.expose()
    def get(self):
        return 'This is RootController GET.'

当存在以上代码时，使用以下命令调用就会有返回值：

curl -X GET http://127.0.0.1:8080

返回值为:

This is RootController GET.

也就是说，curl中的GET对应了代码中的get()方法，若要增加POST方法，可以在代码中添加如下：

@pecan.expose()
def post(self):
    return 'This is RootController POST.'

当然，如果有的方法，需要既可以使用GET，又可以使用POST，且对不同的方法有不同的回应，则可以写成如下：

from pecan import rest
import pecan


class RootController(rest.RestController):
    
    _custom_actions = {
        'test': ['GET', 'POST'],
    }
    
    @pecan.expose()
    def test(self):
        if pecan.request.method == 'POST':
            return 'This is RootController test POST.'
        elif pecan.request.method == 'GET':
            return 'This is RootController test GET.'

可以使用如下命令进行请求：

curl -X GET http://127.0.0.1:8080/test
curl -X POST http://127.0.0.1:8080/test

==下面重点介绍对象分发式路由：==

当存在以下代码的时候：

class v1Controller(rest.RestController):
    @pecan.expose()
    def get(self):
        return 'This is v1Controller GET.'


class RootController(rest.RestController):
    
    v1 = v1Controller()

在RootController类中，生成了一个v1Controller对象，这个对象就是用来做分发式路由的，因此可以使用如下命令去调用它：

curl -X GET http://127.0.0.1:8080/v1

三. 使用WSME来规范API的响应值

wsme模块可以用来规范API的请求和响应值，并且可以和pecan模块很好的结合在一起，其支持的类型如下：

Type Json type
str String
unicode String
int Number
float Number
bool Boolean
Decimal String
date String (YYYY-MM-DD)
time String (hh:mm:ss)
datetime String (YYYY-MM-DDThh:mm:ss)
Arrays Array
None null
Complex types Object

在下面这个例子中，可以看出：

from wsmeext.pecan import wsexpose
from pecan import rest

class RootController(rest.RestController):
    
    @wsexpose(int, int)
    def get_one(self, arg):
        return 1

@wsexpose(int, int)中第一个int表示返回值必须为int，第二个int表示请求参数必须为int

在这个例子中，可以使用curl去测试：

curl -X GET http://127.0.0.1:8081/1

# 返回消息如下：
1

若用以下的输入，则会错误：

curl -X GET http://127.0.0.1:8081/xxx

# 返回消息如下：
{"debuginfo": null, "faultcode": "Client", "faultstring": "Invalid input for field/attribute arg. Value: 'xxx'. unable to convert to int. Error: invalid literal for int() with base 10: 'xxx'"}

当然，wsme还可以用来检测更为复杂的类型，如类对象，这个在下面再做介绍。

四. wsme检测类对象：

在openstack中，使用Rest API返回的响应值经常会是以下格式：

{
    "users": [
        {
            "name": "Alice",
            "age": 30
        }, 
        {
            "name": "Bob", 
            "age": 40
        }
    ]
}

针对于这种情况，我们可以使用WSME的自定义类型，下面就定义一个user类型和users类型：

from wsme import types as wtypes

class User(wtypes.Base):
    name = wtypes.text
    age = int

class Users(wtypes.Base):
    users = [User]

注意： 这里没有使用__init__是因为，父类的初始化方法参数为**kw，因此没有特殊需求，这里可以不写

现在可以使用如下的方法去调用这两个类型：

from wsmeext.pecan import wsexpose
from pecan import rest

class RootController(rest.RestController):
    
	@wsexpose(User, int)		# 返回值为User类对象
    def get_one(self, id):
        if id == 1:
            user_info = {
            	'name': 'yangsijie', 
            	'age': 23
        	}
        return User(**user_info)
    	# 或者也可以使用下面这种方式：
        # if id == 1:
        #     return User(name='yangsijie', age=23)
    
    @wsexpose(Users)	# 返回值为Users类对象
    def get_all(self):
        user_info_list = [
            {
            	'name': 'yangsijie', 
            	'age': 23
        	},
        	{
            	'name': 'panna', 
            	'age': 23
        	}
        ]
        users_list = [User(**user_info) for user_info in user_info_list]
        return Users(users=users_list)
    	# 或者也可以使用下面这种方式：
        # return Users(users=[User(name='yangsijie', age=23), User(name='panna', age=23)])

现在使用curl命令会出现以下现象：

curl http://127.0.0.1:8081/1
# 返回值为：
{"age": 23, "name": "yangsijie"}

curl http://127.0.0.1:8081
# 返回值为：
{"users": [{"age": 23, "name": "yangsijie"}, {"age": 23, "name": "panna"}]}

WSME还可以用于检测上传的参数是否为complex type类型（由于上传要用到POST操作，所以此处就以POST作例子）：

class RootController(rest.RestController):
    
    @wsexpose(None, body=User)	# 检查参数必须为User类对象
    def post(self, user):
        print user.name
        print user.age

当使用curl访问时，程序的控制台会打印出访问的值：

curl -X POST http://127.0.0.1:8081 -H "Content-Type: application/json" -d '{"name": "yangsijie", "age": 30}'

# 程序控制台显示的是：
yangsijie
30

这里需要注意的是，如果这里不写成body=User，而是直接写成User，那么curl中的data字段就也需要进行相应的修改，需要用以下命令来实现：==经测试，这种方法好像根本行不通????==

curl -X POST http://127.0.0.1:8081 -H "Content-Type: application/json" -d '{"user": {"name": "yangsijie", "age": 30}}'

当类中的属性没有传入值的时候，那么这个类变量是wsme.types.UnsetType对象：

from wsme import types as wtyps

user = User(name='test1')	# 这里没有给user对象的age赋值

if user.age is wtypes.Unset:
    return True

结果会返回True，因为此处的age为wsme.types.UnsetType类型。

五. 使用wsme设置status_code

使用wsme默认返回的状态码为200、400和500。

状态码 含义
200 成功
400 客户端输入出错（参数错误等等）
500 服务器端错误

如果将之前的例子做略微的修改，可以让其返回的状态值不一样：

from wsmeext.pecan import wsexpose
from pecan import rest

class RootController(rest.RestController):
    
    @wsexpose(int, int, status_code=201)	# 之前如果成功的话，返回的是200，现在将其改为201
    def get_one(self, arg):
        return 1

在现在这个例子中，使用如上一样的curl命令，会发现返回的状态码变成了201

但是在平时的使用中，我们肯定需要根据不同的判断，返回不同的status_code，那怎么办呢？我们可以使用如下的方式去实现：

  • 使用wsme.api.Response在返回值的同时指定status_code
  • 抛出wsme.exc.ClientSideError错误的同时，指定错误原因及status_code
  • 抛出自定义错误，并且在自定义错误中指定错误原因及status_code

后面两种方式，是遇到错误的时候返回的

下面就看看例子中究竟是如何使用的：

import wsme
from wsme import types as wtypes
from wsmeext.pecan import wsexpose
from pecan import rest


class BookNotFound(Exception):		# 自定义错误
    message = 'Book with ID={id} Not Found'		# 定义错误原因
    code = 404									# 定义status_code

    def __init__(self, id):
        message = self.message.format(id=id)
        super(BookNotFound, self).__init__(message)


class Book(wtypes.Base):
    id = int
    name = wtypes.text


class BookController(rest.RestController):
	@wsexpose(Books, int)
    def get_one(self, id):
        if id == 1:
            raise BookNotFound(id=id)	# 使用自定义错误返回
        elif id == 2:
            raise wsme.exc.ClientSideError('ID: \'1\' is wrong!!!', status_code=403)										# 使用wsme.exc.ClientSideError抛出错误
        else:
            return wsme.api.Response(Books(), status_code=204)																# 使用wsme.api.Response返回值

案例：

该项目实现了用户的查找，添加和删除等（没有真正的结合数据库实现，只是一个demo）

项目的架构图如下：

webdemo2/
├── api
│   ├── app.py					# 存放WSGI application的入口
│   ├── config.py				# 存放Pecan的配置
│   ├── controllers				# 存放Pecan控制器的代码
│   │   ├── __init__.py
│   │   ├── root.py
│   │   └── v1
│   │       ├── controller.py
│   │       ├── __init__.py
│   │       └── users.py
│   ├── expose.py
│   └── __init__.py
├── cmd
│   ├── api.py
│   └── __init__.py
└── __init__.py

本项目实现了以下几个功能：

GET /v1/users 获取所有用户的列表

POST /v1/users 创建一个用户

GET /v1/users/ 获取一个指定用户的详细信息

PUT /v1/users/ 修改一个指定用户的详细信息

DELETE /v1/users/ 删除一个指定用户

POST /v1/users//kill 杀死一个指定用户

api/app.py：

import pecan
from webdemo2.api import config as api_config


def get_pecan_config():
    filename = api_config.__file__.replace('.pyc', '.py')   # get the absolute path of the pecan config.py
    return pecan.configuration.conf_from_file(filename)


def setup_app():      # the main functhing, start listening
    config = get_pecan_config()
    app_conf = dict(config.app)
    app = pecan.make_app(
        app_conf.pop('root'),
        logging=getattr(config, 'logging', {}),
        **app_conf)

    return app

api/config.py:

app = {
    'root': 'webdemo2.api.controllers.root.RootController',
    'modules': ['webdemo2.api'],
    'debug': True,
}

api/expose.py:

import wsmeext.pecan as wsme_pecan


def expose(*args, **kwargs):
    if 'rest_content_types' not in kwargs:
        kwargs['rest_content_types'] = ('json',)
    return wsme_pecan.wsexpose(*args, **kwargs)

该函数用来让API返回JSON格式的数据

api/controllers/root.py:

from pecan import rest, expose
from wsme import types as wtypes
from webdemo2.api.controllers.v1 import controller as v1_controller
from webdemo2.api.expose import expose as wsexpose


class RootController(rest.RestController):

    # All supported API versions
    _versions = ['v1']

    # The default API version
    _default_version = 'v1'

    v1 = v1_controller.V1Controller()

    @wsexpose(wtypes.text)
    def get(self):
        return 'webdemo2'

    @expose()
    def _route(self, args, request=None):
        """When the API version is not specified in the url, v1 is used as the default version."""
        if args[0] and args[0] not in self._versions:
            args = [self._default_version] + args
        return super(RootController, self)._route(args)

当URL中未指定版本号时，_route函数将版本号默认置为v1

api/controllers/v1/controller.py:

from pecan import rest
from wsme import types as wtypes
from webdemo2.api.expose import expose as wsexpose
from webdemo2.api.controllers.v1.users import UsersController


class V1Controller(rest.RestController):

    users = UsersController()

    @wsexpose(wtypes.text)
    def get(self):
        return 'webdemo2 v1controller'

api/controllers/v1/users.py:

from wsme import types as wtypes
from pecan import rest, expose
from webdemo2.api.expose import expose as wsexpose


class User(wtypes.Base):
    id = wtypes.wsattr(wtypes.text, mandatory=True)
    name = wtypes.text
    age = int


class Users(wtypes.Base):
    users = [User]


class UsersController(rest.RestController):

    # HTTP GET /users/
    @wsexpose(Users)
    def get(self):
        user_info_list = [
            {
                'id': '1',
                'name': 'Alice',
                'age': 30
            },
            {
                'id': '2',
                'name': 'Bob',
                'age': 40
            }
        ]
        users_list = [User(**user_info) for user_info in user_info_list]
        return Users(users=users_list)

    # HTTP POST /users
    @wsexpose(None, body=User, status_code=201)
    def post(self, user):
        print user

    @expose()
    def _lookup(self, user_id, *remainder):
        return UserController(user_id), remainder


class UserController(rest.RestController):

    _custom_actions = {
        'kill': ['POST']
    }

    def __init__(self, user_id):
        self.user_id = user_id

    # HTTP GET /users/123456/
    @wsexpose(User)
    def get(self):
        user_info = {
            'id': self.user_id,
            'name': 'Alice',
            'age': 30
        }
        return User(**user_info)

    # HTTP PUT /users/123456/
    @wsexpose(User, body=User)
    def put(self, user):
        user_info = {
            'id': self.user_id,
            'name': user.name,
            'age': user.age + 1
        }
        return User(**user_info)

    # HTTP DELETE /users/123456/
    @wsexpose()
    def delete(self):
        print ('Delete user_id: %s' % self.user_id)

    # HTTP POST /users/123456/kill
    @wsexpose(status_code=202)
    def kill(self):
        print ('Kill user_id: %s' % self.user_id)

cmd/api.py:

from wsgiref import simple_server
from webdemo2.api import app


def main():
    application = app.setup_app()

    srv = simple_server.make_server('', 8081, application)
    print ('Server on port 8081, listening...')

    srv.serve_forever()

    
if __name__ == '__main__':
    main()

运行该文件，即可开始监听

CATALOG
    FEATURED TAGS
    python openstack