高性能的 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, 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()