Falcon FrameworkPython 框架
Falcon 是一个高性能的 Python 框架，用于构建云端 API 和 Web 应用的后端程序。
特性：
通过 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,
href='http://docs.example.com/auth')
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,
href='http://docs.example.com/auth',
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.',
href='http://docs.examples.com/api/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.',
href='http://docs.examples.com/api/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()评论