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