學習Flask

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)

官網flask thread-locals

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)

官網flask request

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)

官網flask response

7. Flask 擴展

官網flask extensions

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)

flask_restful 官網

10. flask_login

@login_manager.user_loader
def load_user(user_id):
    return User.get(user_id)

flask_login 官網

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'})  # 向訂閱者發送參數

flask 信號
blinker 官網