python框架之Flask(1)-Flask初使用

Flask是一个基于 Python 开发并且依赖 jinja2 模板和 Werkzeug WSGI 服务的一个微型框架，对于 Werkzeug 本质是 Socket 服务端，其用于接收 http 请求并对请求进行预处理，然后触发 Flask 框架，开发人员基于 Flask 框架提供的功能对请求进行相应的处理，并返回给用户，如果要返回给用户复杂的内容时，需要借助 jinja2 模板来实现对模板的处理，即：将模板和数据进行渲染，将渲染后的字符串返回给用户浏览器。

准备

安装

pip3 install flask

werkzeug的简单使用

from werkzeug.wrappers import Request, Response
from werkzeug.serving import run_simple


@Request.application
def hello(request):
    return Response('Hello World!')


if __name__ == '__main__':
    run_simple('localhost', 4000, hello)

使用

hello flask

from flask import Flask

app = Flask(__name__)


@app.route('/')
def hello_world():
    return 'Hello World!'


if __name__ == '__main__':
    app.run()

登录示例

from flask import Flask, request, render_template, redirect, session

app = Flask(__name__,
            template_folder='templates',  # 默认模板文件夹
            static_folder='static',  # 默认静态文件文件夹
            static_url_path='/static'  # 默认静态文件访问路径
            )
app.config['DEBUG'] = True
# 使用 session 必须定义 否则报错
app.secret_key = 'A0Zr98j/3yX R~XHH!jmN]LWX/,?RT'


# 默认只支持 GET 请求
@app.route('/login', methods=['GET', "POST"])
def login():
    if request.method == 'GET':
        # 渲染模板 默认对应templates 文件夹
        return render_template('login.html')
    user = request.form.get('user')
    pwd = request.form.get('pwd')

    if user == 'zze' and pwd == '666':
        session['user'] = user
        # 重定向
        return redirect('/index')
    return render_template('login.html', msg='用户名或密码错误！')


@app.route('/index')
def index():
    user = session['user']
    if not user:
        return redirect('/login')
    return render_template('index.html')


if __name__ == '__main__':
    app.run()

配置文件

• 方式一：app.config

  app.config 本质上其实是一个字典，所以可以通过如下方式进行配置：

  from flask import Flask
  
  app = Flask(__name__)
  app.config['DEBUG'] = True

• 方式二：对象配置

  当然，Flask 也提供了单文件配置的方式，如下：

  class Dev:
      DEBUG = True

  from flask import Flask
  
  app = Flask(__name__)
  app.config.from_object('settings.Dev')

• 方式三：文件配置

  app.config.from_pyfile("settings.py")
  # 如：
  # settings.py
  #    DEBUG = True

• 方式四：环境变量

  app.config.from_envvar("环境变量名称")
  # 环境变量的值为python文件名称名称，内部调用from_pyfile方法

• 方式五：json方式

  app.config.from_json("json文件名称")
  # json文件名称，必须是json格式，因为内部会执行json.loads

• 方式六：字典方式

  app.config.from_mapping({'DEBUG':True})
  # 字典格式

• 默认配置参数

  {
          'DEBUG':                                get_debug_flag(default=False),  是否开启Debug模式
          'TESTING':                              False,                          是否开启测试模式
          'PROPAGATE_EXCEPTIONS':                 None,                         
          'PRESERVE_CONTEXT_ON_EXCEPTION':        None,
          'SECRET_KEY':                           None,
          'PERMANENT_SESSION_LIFETIME':           timedelta(days=31),
          'USE_X_SENDFILE':                       False,
          'LOGGER_NAME':                          None,
          'LOGGER_HANDLER_POLICY':               'always',
          'SERVER_NAME':                          None,
          'APPLICATION_ROOT':                     None,
          'SESSION_COOKIE_NAME':                  'session',
          'SESSION_COOKIE_DOMAIN':                None,
          'SESSION_COOKIE_PATH':                  None,
          'SESSION_COOKIE_HTTPONLY':              True,
          'SESSION_COOKIE_SECURE':                False,
          'SESSION_REFRESH_EACH_REQUEST':         True,
          'MAX_CONTENT_LENGTH':                   None,
          'SEND_FILE_MAX_AGE_DEFAULT':            timedelta(hours=12),
          'TRAP_BAD_REQUEST_ERRORS':              False,
          'TRAP_HTTP_EXCEPTIONS':                 False,
          'EXPLAIN_TEMPLATE_LOADING':             False,
          'PREFERRED_URL_SCHEME':                 'http',
          'JSON_AS_ASCII':                        True,
          'JSON_SORT_KEYS':                       True,
          'JSONIFY_PRETTYPRINT_REGULAR':          True,
          'JSONIFY_MIMETYPE':                     'application/json',
          'TEMPLATES_AUTO_RELOAD':                None,
  }

路由系统

@app.route('/user/<username>')
@app.route('/post/<int:post_id>')
@app.route('/post/<float:post_id>')
@app.route('/post/<path:path>')
@app.route('/login', methods=['GET', 'POST'])

常用路由系统有以上五种，所有的路由系统都是基于一下对应关系来处理：

DEFAULT_CONVERTERS = {
    'default':          UnicodeConverter,
    'string':           UnicodeConverter,
    'any':              AnyConverter,
    'path':             PathConverter,
    'int':              IntegerConverter,
    'float':            FloatConverter,
    'uuid':             UUIDConverter,
}

可以通过 endpoint 和 url_for 反向生成：

from flask import Flask, url_for

app = Flask(__name__)


@app.route('/index', endpoint='i')
def index():
    return url_for('i')  # endpoint 值未指定时，默认为方法名


if __name__ == '__main__':
    app.run()

模板

• 语法

  Flask 使用的是 Jinjia2 模板，所以其语法和在 Django 中使用时无差别。

  Markup 等价 django 的 mark_safe。

• 模板方法

  {% macro input(name, type='text', value='') %}
      <input type="{{ type }}" name="{{ name }}" value="{{ value }}">
  {% endmacro %}
  
  {{ input('n1') }}

请求和响应

• 请求相关

  request 内置属性：
      request.method # 方法
      request.args # GET 请求时的参数
      request.form # POST 请求时的参数
      request.values # 所有参数
      request.cookies # cookie 内容
      request.headers # 请求头信息
      request.path # 请求路径
      request.full_path # 请求全路径
  
      路径相关：
          request.script_root
          request.url
          request.base_url
          request.url_root
          request.host_url
          request.host
      
      request.files # 上传文件接收 

  # 保存文件
  obj = request.files['the_file_name']
  obj.save('/var/www/uploads/' + secure_filename(f.filename))

• 响应相关

  # 返回字符串
  return "字符串"
  # 返回渲染模板
  return render_template('html模板路径',**{})
  # 重定向
  return redirect('/index.html')

Session

当请求刚到来时，flask 读取 cookie 中 session 对应的值，并将该值解密并反序列化成字典，放入内存以便视图函数使用。

当请求结束时，flask 会读取内存中字典的值，进行序列化和加密，再写入到 cookie 中。

中间件

从 flask 的入口 app.run 方法看起，

def run(self, host=None, port=None, debug=None,
        load_dotenv=True, **options):
    if os.environ.get('FLASK_RUN_FROM_CLI') == 'true':
        from .debughelpers import explain_ignored_app_run
        explain_ignored_app_run()
        return

    if get_load_dotenv(load_dotenv):
        cli.load_dotenv()

        # if set, let env vars override previous values
        if 'FLASK_ENV' in os.environ:
            self.env = get_env()
            self.debug = get_debug_flag()
        elif 'FLASK_DEBUG' in os.environ:
            self.debug = get_debug_flag()

    if debug is not None:
        self.debug = bool(debug)

    _host = '127.0.0.1'
    _port = 5000
    server_name = self.config.get('SERVER_NAME')
    sn_host, sn_port = None, None

    if server_name:
        sn_host, _, sn_port = server_name.partition(':')

    host = host or sn_host or _host
    port = int(port or sn_port or _port)

    options.setdefault('use_reloader', self.debug)
    options.setdefault('use_debugger', self.debug)
    options.setdefault('threaded', True)

    cli.show_server_banner(self.env, self.debug, self.name, False)

    from werkzeug.serving import run_simple

    try:
        run_simple(host, port, self, **options)
    finally:
        self._got_first_request = False

直接看到 41 行，这里的 run_simple 方法实际上就是 werkzeug.serving.run_simple ，而传入的第三个参数 self 实际上就是指的 app 本身。我们先要了解的是，在上面的werkzeug的简单使用中，传入的是一个方法，并且这个方法会在请求到来时被执行。而在这里传入一个对象，加 () 执行一个对象时实际上是执行这个对象的 __call__ 方法，查看 app.__call__ 方法：

def __call__(self, environ, start_response):
    return self.wsgi_app(environ, start_response)

看到这里我们可以确定，后续整个程序的执行就是依赖这第 2 行的触发，而我们只要在第二行执行之前定义的处理就相当于中间件的前置处理，在它之后的处理就相当于后置处理了。所以可以通过如下方法实现中间件的功能：

from flask import Flask

app = Flask(__name__)


class MiddleWare:
    def __init__(self, wsgi_app):
        self.wsgi_app = wsgi_app

    def __call__(self, *args, **kwargs):
        print('前置处理')
        obj = self.wsgi_app(*args, **kwargs)
        print('后置处理')


if __name__ == "__main__":
    app.wsgi_app = MiddleWare(app.wsgi_app)
    app.run(port=9999)

flash

flash 就是用来存储只使用一次的数据，类似于将数据放入列表，然后通过 pop 方法取出。

flash(message, category='message')
get_flashed_messages(with_categories=False, category_filter=[])

• 使用

  from flask import Flask, flash, get_flashed_messages
  
  app = Flask(__name__)
  
  
  @app.route('/page1')
  def page1():
      flash('临时数据')
      return 'page1'
  
  
  @app.route('/page2')
  def page2():
      get_flashed_messages()
      return 'page2'
  
  
  if __name__ == '__main__':
      app.run()

• 源码分析

  查看 flash 方法：

  def flash(message, category='message'):
      flashes = session.get('_flashes', [])
      flashes.append((category, message))
      session['_flashes'] = flashes
      message_flashed.send(current_app._get_current_object(),
                           message=message, category=category)

  可以看到它其实就是把 message 和 category 以元组的形式存储在 session 中键为 _flashes 的值中。

  再看 get_flashed_messages 方法：

  def get_flashed_messages(with_categories=False, category_filter=[]):
      flashes = _request_ctx_stack.top.flashes
      if flashes is None:
          _request_ctx_stack.top.flashes = flashes = session.pop('_flashes') 
              if '_flashes' in session else []
      if category_filter:
          flashes = list(filter(lambda f: f[0] in category_filter, flashes))
      if not with_categories:
          return [x[1] for x in flashes]
      return flashes

  从第 4 行可以看到每次获取 flash 数据时，就是从 session 中将之前存入的键为 _flashes 的元素 pop 出来，然后进行过滤返回对应 category 内容。

特殊的装饰器

• before_request&after_request

  from flask import Flask, flash, get_flashed_messages
  
  app = Flask(__name__)
  app.secret_key = '123123'
  
  
  @app.before_request
  def before_request1():
      print('before_request1')
  
  
  @app.before_request
  def before_request2():
      print('before_request2')
  
  
  @app.after_request
  def after_request1(response):
      print('after_request1')
      return response
  
  
  @app.after_request
  def after_request2(response):
      print('after_request2')
      return response
  
  
  @app.route('/index')
  def index():
      print('index')
      return 'index'
  
  
  if __name__ == '__main__':
      app.run()
      
  # 执行顺序如下：
  #     before_request1
  #     before_request2
  #     index
  #     after_request2
  #     after_request1

• template_filter

  from flask import Flask, flash, get_flashed_messages, render_template
  
  app = Flask(__name__)
  app.secret_key = '123123'
  
  
  @app.template_filter()
  def add_filter(a, b):
      return a + b
  
  
  @app.route('/index')
  def index():
      print('index')
      return render_template('index.html')
  
  
  if __name__ == '__main__':
      app.run()

  <!DOCTYPE html>
  <html lang="en">
  <head>
      <meta charset="UTF-8">
      <title>Title</title>
  </head>
  <body>
  {{1|add_filter(2)}}
  </body>
  </html>
  

• template_global

  from flask import Flask, flash, get_flashed_messages, render_template
  
  app = Flask(__name__)
  app.secret_key = '123123'
  
  
  @app.template_global()
  def add_template_func(a, b):
      return a + b
  
  
  @app.route('/index')
  def index():
      print('index')
      return render_template('index.html')
  
  
  if __name__ == '__main__':
      app.run()

  <!DOCTYPE html>
  <html lang="en">
  <head>
      <meta charset="UTF-8">
      <title>Title</title>
  </head>
  <body>
  {{add_template_func(1,2)}}
  </body>
  </html>
  

• errorhandler

  @app.errorhandler(404)
  def page_not_found(error):
      return render_template('404.html'),404

• before_first_request

  @app.before_first_request
  def application_start():
      # 同 before_request，但仅在第一次请求时执行一次。
      pass