class JSONPlugin(object):
name = 'json'
api = 2
def __init__(self, json_dumps=json_dumps):
self.json_dumps = json_dumps
def setup(self, app):
app.config._define('json.enable', default=True, validate=bool,
help="Enable or disable automatic dict->json filter.")
app.config._define('json.ascii', default=False, validate=bool,
help="Use only 7-bit ASCII characters in output.")
app.config._define('json.indent', default=True, validate=bool,
help="Add whitespace to make json more readable.")
app.config._define('json.dump_func', default=None,
help="If defined, use this function to transform"
" dict into json. The other options no longer"
" apply.")
def apply(self, callback, route):
dumps = self.json_dumps
if not self.json_dumps: return callback
def wrapper(*a, **ka):
try:
rv = callback(*a, **ka)
except HTTPResponse as resp:
rv = resp
if isinstance(rv, dict):
#Attempt to serialize, raises exception on failure
json_response = dumps(rv)
#Set content type only if serialization successful
response.content_type = 'application/json'
return json_response
elif isinstance(rv, HTTPResponse) and isinstance(rv.body, dict):
rv.body = dumps(rv.body)
rv.content_type = 'application/json'
return rv
return wrapper
class TemplatePlugin(object):
""" This plugin applies the :func:`view` decorator to all routes with a
`template` config parameter. If the parameter is a tuple, the second
element must be a dict with additional options (e.g. `template_engine`)
or default variables for the template. """
name = 'template'
api = 2
def setup(self, app):
app.tpl = self
def apply(self, callback, route):
conf = route.config.get('template')
if isinstance(conf, (tuple, list)) and len(conf) == 2:
return view(conf[0], **conf[1])(callback)
elif isinstance(conf, str):
return view(conf)(callback)
else:
return callback
def make_middleware_decorator(middleware_class):
def _make_decorator(*m_args, **m_kwargs):
middleware = middleware_class(*m_args, **m_kwargs)
def _decorator(view_func):
@wraps(view_func, assigned=available_attrs(view_func))
def _wrapped_view(request, *args, **kwargs):
# Request 预处理函数
if hasattr(middleware, 'process_request'):
result = middleware.process_request(request)
if result is not None:
return result
# View 预处理函数
if hasattr(middleware, 'process_view'):
result = middleware.process_view(request, view_func, args, kwargs)
if result is not None:
return result
try:
response = view_func(request, *args, **kwargs)
except Exception as e:
# Exception 后处理函数
if hasattr(middleware, 'process_exception'):
result = middleware.process_exception(request, e)
if result is not None:
return result
raise
if hasattr(response, 'render') and callable(response.render):
# Template 模版渲染函数
if hasattr(middleware, 'process_template_response'):
response = middleware.process_template_response(request, response)
# Defer running of process_response until after the template
# has been rendered:
# Response 后处理函数
if hasattr(middleware, 'process_response'):
def callback(response):
return middleware.process_response(request, response)
response.add_post_render_callback(callback)
else:
# Response 后处理函数
if hasattr(middleware, 'process_response'):
return middleware.process_response(request, response)
return response
return _wrapped_view
return _decorator
return _make_decorator
from werkzeug.wrappers import Request, Response, ResponseStream
class middleware():
'''
Simple WSGI middleware
'''
def __init__(self, app):
self.app = app
self.userName = 'Tony'
self.password = 'IamIronMan'
def __call__(self, environ, start_response):
request = Request(environ)
userName = request.authorization['username']
password = request.authorization['password']
# these are hardcoded for demonstration
# verify the username and password from some database or env config variable
if userName == self.userName and password == self.password:
environ['user'] = { 'name': 'Tony' }
return self.app(environ, start_response)
res = Response(u'Authorization failed', mimetype= 'text/plain', status=401)
return res(environ, start_response)
server.py
from flask import Flask, request
from middleware import middleware
app = Flask('DemoApp')
# calling our middleware
app.wsgi_app = middleware(app.wsgi_app)
@app.route('/', methods=['GET', 'POST'])
def hello():
# using
user = request.environ['user']
return "Hi %s"%user['name']
if __name__ == "__main__":
app.run('127.0.0.1', '5000', debug=True)
2.3.3.2 源码解读
class Flask(Scaffold):
...
def wsgi_app(self, environ: dict, start_response: t.Callable) -> t.Any:
"""The actual WSGI application. This is not implemented in
:meth:`__call__` so that middlewares can be applied without
losing a reference to the app object. Instead of doing this::
app = MyMiddleware(app)
It's a better idea to do this instead::
app.wsgi_app = MyMiddleware(app.wsgi_app)
Then you still have the original application object around and
can continue to call methods on it.
.. versionchanged:: 0.7
Teardown events for the request and app contexts are called
even if an unhandled error occurs. Other events may not be
called depending on when an error occurs during dispatch.
See :ref:`callbacks-and-errors`.
:param environ: A WSGI environment.
:param start_response: A callable accepting a status code,
a list of headers, and an optional exception context to
start the response.
"""
ctx = self.request_context(environ)
error: t.Optional[BaseException] = None
try:
try:
ctx.push()
response = self.full_dispatch_request()
except Exception as e:
error = e
response = self.handle_exception(e)
except: # noqa: B001
error = sys.exc_info()[1]
raise
return response(environ, start_response)
finally:
if self.should_ignore_error(error):
error = None
ctx.auto_pop(error)
def __call__(self, environ: dict, start_response: t.Callable) -> t.Any:
"""The WSGI server calls the Flask application object as the
WSGI application. This calls :meth:`wsgi_app`, which can be
wrapped to apply middleware.
"""
return self.wsgi_app(environ, start_response)
from __future__ import absolute_import
import uuid
from webob.headers import ResponseHeaders
from st2common.constants.api import REQUEST_ID_HEADER
from st2common.router import Request
class RequestIDMiddleware(object):
def __init__(self, app):
self.app = app
def __call__(self, environ, start_response):
# The middleware adds unique `X-Request-ID` header on the requests that don't have it and
# modifies the responses to have the same exact header as their request. The middleware
# helps us better track relation between request and response in places where it might not
# be immediately obvious (like logs for example). In general, you want to place this header
# as soon as possible to ensure it's present by the time it's needed. Certainly before
# LoggingMiddleware which relies on this header.
request = Request(environ)
if not request.headers.get(REQUEST_ID_HEADER, None):
req_id = str(uuid.uuid4())
request.headers[REQUEST_ID_HEADER] = req_id
def custom_start_response(status, headers, exc_info=None):
headers = ResponseHeaders(headers)
req_id_header = request.headers.get(REQUEST_ID_HEADER, None)
if req_id_header:
headers[REQUEST_ID_HEADER] = req_id_header
return start_response(status, headers._items, exc_info)
return self.app(environ, custom_start_response)
4.2.4.2.1 关闭 db 连接 -- （使用数据库时，建议使用自动重连配置，则无需再添加关闭 db 连接配置）
from xlib.db import my_database
from xlib.util import wrapt
@wrapt.decorator
def db_close(wrapped, instance, args, kwargs):
result = wrapped(*args,**kwargs)
if not my_database.is_closed():
my_database.close()
return result
4.2.4.2.2 检查是否已登录 -- （使用『火眼』组件，无需通过装饰器再实现）
from xlib import auth
@wrapt.decorator
def login_required(wrapped, instance, args, kwargs):
"""
使用 web 接口访问时，是通过 kwargs 进行传值
使用 test_handler.py 进行接口调试时，则是使用 args 进行传值，所以接口调试时会自动绕过身份验证
"""
if "req" in kwargs:
req = kwargs["req"]
token_check = auth.is_token_valid(req)
if not token_check.success:
return token_check.err_content
result = wrapped(*args,**kwargs)
return result
