高性能的 Python 框架：Falcon Framework

Falcon 是一個高性能的 Python 框架，用于構建云端 API 和 Web 應用的后端程序。

設計目標：

Fast. Cloud APIs need to turn around requests quickly, and make efficient use of hardware. This is particularly important when serving many concurrent requests. Falcon processes requests several times faster than other popular web frameworks.

Light. Only the essentials are included, with six and mimeparse being the only dependencies outside the standard library. We work to keep the code lean, making Falcon easier to test, optimize, and deploy.

Flexible. Falcon can be deployed in a variety of ways, depending on your needs. The framework speaks WSGI, and works great with Python 2.6 and 2.7, PyPy, and Python 3.3/3.4. There's no tight coupling with any async framework, leaving you free to mix-and-match what you need.

特性：

• 通過 URI 模板和資源類可直觀的了解路由信息

• 輕松訪問請求和響應類來訪問 header 和 body 信息

• 通過方便的異常類實現對 HTTP 錯誤響應的處理

• 通過全局、資源和方法鉤子實現 DRY 請求處理

• 通過 WSGI helper 和 mock 實現單元測試

• 使用 Cython 可提升 20% 的速度

• 支持 Python 2.6, Python 2.7, PyPy 和 Python 3.3/3.4

• 高性能！！！

一個比較完整的例子：

import json
import logging
import uuid
from wsgiref import simple_server

import falcon


class StorageEngine(object):
    def get_things(self, marker, limit):
        return []

    def add_thing(self, thing):
        return {'id': str(uuid.uuid4())}


class StorageError(Exception):
    @staticmethod
    def handle(ex, req, resp, params):
        description = ('Sorry, couldn\'t write your thing to the '
                       'database. It worked on my box.')

        raise falcon.HTTPError(falcon.HTTP_725,
                               'Database Error',
                               description)


class Proxy(object):
    def forward(self, req):
        return falcon.HTTP_503


class SinkAdapter(object):

    def __init__(self):
        self._proxy = Proxy()

    def __call__(self, req, resp, **kwargs):
        resp.status = self._proxy.forward(req)
        self.kwargs = kwargs


def token_is_valid(token, user_id):
    return True  # Suuuuuure it's valid...


def auth(req, resp, params):
    # Alternatively, use Talons or do this in WSGI middleware...
    token = req.get_header('X-Auth-Token')

    if token is None:
        description = ('Please provide an auth token '
                       'as part of the request.')

        raise falcon.HTTPUnauthorized('Auth token required',
                                      description,
                                      )

    if not token_is_valid(token, params['user_id']):
        description = ('The provided auth token is not valid. '
                       'Please request a new token and try again.')

        raise falcon.HTTPUnauthorized('Authentication required',
                                      description,
                                      ,
                                      scheme='Token; UUID')


def check_media_type(req, resp, params):
    if not req.client_accepts_json:
        raise falcon.HTTPNotAcceptable(
            'This API only supports responses encoded as JSON.',
            )

    if req.method in ('POST', 'PUT'):
        if not req.content_type == 'application/json':
            raise falcon.HTTPUnsupportedMediaType(
                'This API only supports requests encoded as JSON.',
                )


def deserialize(req, resp, resource, params):
    # req.stream corresponds to the WSGI wsgi.input environ variable,
    # and allows you to read bytes from the request body.
    #
    # See also: PEP 3333
    body = req.stream.read()
    if not body:
        raise falcon.HTTPBadRequest('Empty request body',
                                    'A valid JSON document is required.')

    try:
        params['doc'] = json.loads(body.decode('utf-8'))

    except (ValueError, UnicodeDecodeError):
        raise falcon.HTTPError(falcon.HTTP_753,
                               'Malformed JSON',
                               'Could not decode the request body. The '
                               'JSON was incorrect or not encoded as UTF-8.')


def serialize(req, resp, resource):
    resp.body = json.dumps(req.context['doc'])


class ThingsResource:

    def __init__(self, db):
        self.db = db
        self.logger = logging.getLogger('thingsapp.' + __name__)

    @falcon.after(serialize)
    def on_get(self, req, resp, user_id):
        marker = req.get_param('marker') or ''
        limit = req.get_param_as_int('limit') or 50

        try:
            result = self.db.get_things(marker, limit)
        except Exception as ex:
            self.logger.error(ex)

            description = ('Aliens have attacked our base! We will '
                           'be back as soon as we fight them off. '
                           'We appreciate your patience.')

            raise falcon.HTTPServiceUnavailable(
                'Service Outage',
                description,
                30)

        # An alternative way of doing DRY serialization would be to
        # create a custom class that inherits from falcon.Request. This
        # class could, for example, have an additional 'doc' property
        # that would serialize to JSON under the covers.
        req.context['doc'] = result

        resp.set_header('X-Powered-By', 'Small Furry Creatures')
        resp.status = falcon.HTTP_200

    @falcon.before(deserialize)
    def on_post(self, req, resp, user_id, doc):
        proper_thing = self.db.add_thing(doc)

        resp.status = falcon.HTTP_201
        resp.location = '/%s/things/%s' % (user_id, proper_thing['id'])


# Configure your WSGI server to load "things.app" (app is a WSGI callable)
app = falcon.API(before=[auth, check_media_type])

db = StorageEngine()
things = ThingsResource(db)
app.add_route('/{user_id}/things', things)

# If a responder ever raised an instance of StorageError, pass control to
# the given handler.
app.add_error_handler(StorageError, StorageError.handle)

# Proxy some things to another service; this example shows how you might
# send parts of an API off to a legacy system that hasn't been upgraded
# yet, or perhaps is a single cluster that all data centers have to share.
sink = SinkAdapter()
app.add_sink(sink, r'/v1/[charts|inventory]')

# Useful for debugging problems in your API; works with pdb.set_trace()
if __name__ == '__main__':
    httpd = simple_server.make_server('127.0.0.1', 8000, app)
    httpd.serve_forever()

項目主頁：http://www.baiduhome.net/lib/view/home/1420461101516