1. WSGI协议
def application(environ, start_response):
start_response('200 OK', [('Content-Type', 'text/plain')])
return [b'Hello world']
if __name__ == "__main__":
from werkzeug.serving import run_simple
run_simple("localhost", 5000, application)
- start_response(status, response_headers)
status --> str
response --> a list of (header_name, header_value) tuples describing the HTTP response header - environ --> dict
- return --> an iterable yielding zero or more bytestrings
参考PEP3333
2.Flask应用符合WSGI协议
class Flask(...):
...
def __call__(self, environ, start_response):
"""The WSGI server calls the Flask application object as the
WSGI application. This calls :meth:`wsgi_app` which can be
wrapped to applying middleware."""
return self.wsgi_app(environ, start_response)
def wsgi_app(self, environ, start_response):
"""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 = None
try:
try:
ctx.push()
response = self.full_dispatch_request()
except Exception as e:
error = e
response = self.handle_exception(e)
except: # 这里会捕获exit
error = sys.exc_info()[1]
raise
return response(environ, start_response)
finally:
if self.should_ignore_error(error):
error = None
ctx.auto_pop(error)
参考Flask源码
3. 应用上下文和请求上下文
class RequestContext(...):
...
def push(self):
"""Binds the request context to the current context."""
# If an exception occurs in debug mode or if context preservation is
# activated under exception situations exactly one context stays
# on the stack. The rationale is that you want to access that
# information under debug situations. However if someone forgets to
# pop that context again we want to make sure that on the next push
# it's invalidated, otherwise we run at risk that something leaks
# memory. This is usually only a problem in test suite since this
# functionality is not active in production environments.
top = _request_ctx_stack.top
if top is not None and top.preserved:
top.pop(top._preserved_exc)
# Before we push the request context we have to ensure that there
# is an application context.
app_ctx = _app_ctx_stack.top
if app_ctx is None or app_ctx.app != self.app:
app_ctx = self.app.app_context()
app_ctx.push()
self._implicit_app_ctx_stack.append(app_ctx)
else:
self._implicit_app_ctx_stack.append(None)
if hasattr(sys, 'exc_clear'):
sys.exc_clear()
_request_ctx_stack.push(self)
# Open the session at the moment that the request context is available.
# This allows a custom open_session method to use the request context.
# Only open a new session if this is the first time the request was
# pushed, otherwise stream_with_context loses the session.
if self.session is None:
session_interface = self.app.session_interface
self.session = session_interface.open_session(
self.app, self.request
)
if self.session is None:
self.session = session_interface.make_null_session(self.app)
5. Flask Request
>>> from werkzeug.wrappers import Request
>>> request = Request(environ)
from flask import request # instance
request = LocalProxy(partial(_lookup_req_object, 'request'))
def _lookup_req_object(name):
top = _request_ctx_stack.top
if top is None:
raise RuntimeError(_request_ctx_err_msg)
return getattr(top, name)
6. Flask Response
>>> from werkzeug.wrappers import Response
>>> response = Response("Hello World!")
>>> response.headers['content-type']
'text/plain; charset=utf-8'
>>> response.data
'Hello World!'
>>> response.headers['content-length'] = len(response.data)
7. Flask 扩展
from flask_foo import Foo
foo = Foo()
app = Flask(__name__)
app.config.update(
FOO_BAR='baz',
FOO_SPAM='eggs',
)
foo.init_app(app)
8. webargs
- 单字段验证
param_args = {
'plat': Arg(str, default=None, validate=validate_plat_all),
'cid': Arg(str, default=None),
'bid': Arg(str, default=None)
}
@app.route("/register", methods=["POST"])
@use_kwargs(param_args)
def register(args):
return register_user(
args["username"],
args["password"],
fullname=args["fullname"],
per_page=args["display_per_page"],
)
- 多字段验证
@use_kwargs(param_args, validate=validate_all)
9. flask_restful
from flask import Flask
from flask_restful import Resource, Api
app = Flask(__name__)
api = Api(app)
class HelloWorld(Resource):
def get(self):
return {'hello': 'world'}
api.add_resource(HelloWorld, '/')
if __name__ == '__main__':
app.run(debug=True)
10. flask_login
@login_manager.user_loader
def load_user(user_id):
return User.get(user_id)
11. flask 信号
from blinker import signal
do_sth = signal('do_sth') #创建信号
def process(f, a, b, **kwargs):
f(a, b, **kwargs)
print('Has done!')
def f(a, b, **kwargs):
print(f'loc: {a}, {b}; kwargs:{kwargs}')
print('do something.')
do_sth.connect(process) # process订阅信号
do_sth.send(f, **{'hello': 'world', 'a': 'a', 'b': 'b'}) # 向订阅者发送参数