Flask官方文档学习笔记

Posted by Waynerv on

category: Web开发

Tags: Flask

前言

  1. Flask是一个微型的框架,仅提供核心的功能,其他功能依赖于各种第三方拓展
  2. Flask在内部使用了本地线程对象,因此你不用考虑在内部的函数间传递对象,但应注意请求上下文和程序上下文
  3. 开发Web应用应始终考虑安全问题

安装

依赖包

内置

  • Werkzeug 用于实现 WSGI ,应用和服务之间的标准 Python 接口。
  • Jinja 用于渲染页面的模板语言。
  • MarkupSafe 与 Jinja共用,在渲染页面时用于避免不可信的输入,防止注入攻击。
  • ItsDangerous 保证数据完整性的安全标志数据,用于保护 Flask 的 session cookie.
  • Click 是一个命令行应用的框架。用于提供flask命令,并允许添加自定义管理命令。

非内置

  • Blinker 为 信号 提供支持。
  • SimpleJSON 是一个快速的 JSON 实现,兼容 Python’s json模块。如果安装 了这个软件,那么会优先使用这个软件来进行 JSON操作。
  • python-dotenv 当运行 flask 命令时为 通过 dotenv 设置环境变量提供支持。
  • Watchdog 为开发服务器提供快速高效的重载。

虚拟环境

  1. Python3使用venv,Python2使用virtualenv

快速入门

一个最小的应用

from flask import Flask
app = Flask(__name__)

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

$ export FLASK_APP=hello.py
$ flask run
 * Running on http://127.0.0.1:5000/

windows系统下设置启动app环境变量(linux为export):

C:\path\to\app>set FLASK_APP=hello.py

设置服务器外部可见:

flask run --host=0.0.0.0

调试模式

  1. 开启调试模式:设置环境变量FLASK_ENV=development
  2. 开启后激活调试器、激活自动重载并打开 Flask 应用的调试模式
  3. 调试模式必须仅用于开发环境!

路由

变量规则

  1. 通过把URL的一部分标记为就可以在URL中添加变量,端点函数接受变量作为关键字参数,使用,限定变量的数据类型
  2. Converter 类型:
类型 限定规则
string (default) accepts any text without a slash
int accepts positive integers
float accepts positive floating point values
path like string but also accepts slashes
uuid accepts UUID strings
  1. 示例
@app.route('/user/<username>')
def show_user_profile(username):
    # 为指定的user展示用户资料
    return 'User %s' % username

@app.route('/post/<int:post_id>')
def show_post(post_id):
    return 'Post %d' % post_id

唯一的URL / 重定向行为

  1. 端点URL若以/结尾,访问时不加斜杠会被Flask自动重定向到该页面
  2. 端点URL若不以/结尾,访问时加斜杠会404(类似于文件和文件夹)

URL构建

  1. url_for()函数用于构建指定函数的URL,端点名称作为第一个参数,接受任意个关键字参数对应 URL 中的变量,未知变量将添加到 URL 中作为查询参数。
  2. 使用url_for函数的好处:
    • 反向解析通常比直接硬编码 URL 的描述性更好。
    • 你可以只需一个地方改变 URL ,而不用到处手动修改。
    • URL 创建会为你处理特殊字符的转义和 Unicode 数据,比较直观。
    • 创建的路径总是绝对路径,可以避免相对路径产生副作用。
    • 如果你的应用是放在 URL 根路径之外的地方(如在 /myapplication 中,不在/中),url_for() 会为你妥善处理。

HTTP方法

  1. Flask路由默认只处理GET请求,需要使用route()装饰器的methods参数来处理不同的HTTP方法
@app.route('/login', methods=['GET', 'POST'])

静态文件

使用特定的 'static' 端点就可以生成相应的 URL

url_for('static', filename='style.css')

这个静态文件在文件系统中的位置应该是 static/style.css 。

模板

  1. flask使用jinja2引擎来渲染模板
  2. 在模板内部可以访问 get_flashed_messages() 函数、request 、session 和 g 对象。
from flask import render_template

@app.route('/hello/')
@app.route('/hello/<name>')
def hello(name=None):
    return render_template('hello.html', name=name)

访问请求数据

本地环境

  1. 某些对象在 Flask 中是全局对象,但不是通常意义下的全局对象。这些对象实际上是 特定环境下本地对象的代理。
  2. 单元测试时的对策:创建一个请求对象并绑定到环境
    • 使用 test_request_context() 环境管理器。通过使用 with 语句可以绑定一个测试请求,以便于交互。 with app.test_request_context('/hello', method='POST'):
    • 整个 WSGI 环境传递给 request_context() 方法 with app.request_context(environ):

请求对象

  1. 请求的中数据保存在request对象中
  2. 使用 form 属性处理表单数据(在POST或者PUT请求中传输的数据)(以字典形式在form中保存)
  3. 要操作 URL (如 ?key=value )中提交的参数可以使用 args 属性
searchword = request.args.get('key', '')

上传文件

  1. 确保在HTML表单中设置enctype="multipart/form-data"属性。否则浏览器将不会传送文件
  2. 上传的文件保存在内存或临时文件存储位置中,可以通过查找request对象files 属性来获取,也是以字典形式保存:f = request.files['the_file']
  3. 上传的文件可以通过save()方法存储到服务器的指定位置
  4. 注意在保存用户上传文件时对文件名.filename进行转义或secure_filename()处理,否则存在安全风险
from flask import request
from werkzeug.utils import secure_filename

@app.route('/upload', methods=['GET', 'POST'])
def upload_file():
    if request.method == 'POST':
        f = request.files['the_file']
        # f.save('/var/www/uploads/uploaded_file.txt')
        f.save('/var/www/uploads/' + secure_filename(f.filename))

Cookie

  1. request对象的cookies属性也是一个包含了客户端所传输的所有cookies的字典结构
  2. 为确保安全性使用sessions,尽量不直接使用cookies
  3. 读取cookies:
from flask import request

@app.route('/')
def index():
    username = request.cookies.get('username')
    # use cookies.get(key) instead of cookies[key] to not get a KeyError if the cookie is missing.
  1. 设置cookies(设置在response对象中):
from flask import make_response

@app.route('/')
def index():
    resp = make_response(render_template(...))
    resp.set_cookie('username', 'the username')
    return resp

重定向和错误

  1. 使用redirect() 函数可以重定向到其他URL。用abort()函数可以提前中断一个请求并带有一个错误代码。
from flask import abort, redirect, url_for

@app.route('/')
def index():
    return redirect(url_for('login'))

@app.route('/login')
def login():
    abort(401)
    this_is_never_executed()
  1. 使用error_handler()装饰器自定义出错页面:
from flask import render_template

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

关于响应

  1. Flask会将视图函数的返回值会自动转换为一个响应对象。
  2. 如果返回值是字符串,那么会被转换为一个包含:作为响应体的字符串、一个 200 OK 状态代码和一个text/html mimetype首部的响应对象
  3. 具体的转换规则:
    • 如果视图返回的是一个响应对象,那么直接返回它。
    • 如果返回的是一个字符串,那么根据这个字符串和缺省参数生成一个用于返回的 响应对象。
    • 如果返回的是一个元组,元组中必须由(response, status, headers) 或者(response, headers)组成。status的值会重载状态代码,headers是一个由额外头部值组成的列表或字典。
    • 如果以上都不是,那么Flask会假定返回值是一个有效的WSGI应用并把它转换为一个响应对象。
  4. 使用make_response()自行定义返回结果:
@app.errorhandler(404)
def not_found(error):
    # 把原return的对象用make_response()包
    resp = make_response(render_template('error.html'), 404) 装
    resp.headers['X-Something'] = 'A value'  # 自定义响应首部
    return resp

会话Sessions

  1. 使用会话之前必须设置一个密钥:app.secret_key = b'_5#y2L"F4Q8z\n\xec]/'
  2. 快捷生成随机密钥:$ python -c 'import os; print(os.urandom(16))'

消息闪现

  1. flash() 用于闪现一个消息。在模板中,使用 get_flashed_messages() 来操作消息

教程

项目布局

  1. 首先设置虚拟环境(pipenv)
  2. 创建项目目录,通用布局如下:
/home/user/Projects/flask-tutorial
├── flaskr/
│   ├── __init__.py
│   ├── db.py
│   ├── schema.sql
│   ├── auth.py
│   ├── blog.py
│   ├── templates/
│   │   ├── base.html
│   │   ├── auth/
│   │   │   ├── login.html
│   │   │   └── register.html
│   │   └── blog/
│   │       ├── create.html
│   │       ├── index.html
│   │       └── update.html
│   └── static/
│       └── style.css
├── tests/
│   ├── conftest.py
│   ├── data.sql
│   ├── test_factory.py
│   ├── test_db.py
│   ├── test_auth.py
│   └── test_blog.py
├── venv/
├── setup.py
└── MANIFEST.in
  1. 如果使用git,应该设置.gitignore文件忽略项目产生的临时文件(基本原则:不是自己创建的文件就可以忽略)

应用设置

  1. 模块化的项目中逐个手动创建实例是非常低效的,因此用应用工厂函数创建 Flask 实例来代替创建全局实例

应用工厂

  1. 在__init__.py 中创建应用工厂;同时表示Python flaskr 文件夹应当视作为一个包
  2. 在应用工厂中创建实例,设置缺省配置,创建实例文件夹
import os
from flask import Flask

def create_app(test_config=None):
    app = Flask(__name__, instance_relative_config=True)
    app.config.from_mapping(
        SECRET_KEY='dev',
        DATABASE=os.path.join(app.instance_path, 'flaskr.sqlite'),
    )

    if test_config is None:
        app.config.from_pyfile('config.py', silent=True)
    else:
        app.config.from_mapping(test_config)

    try:
        os.makedirs(app.instance_path)
    except OSError:
        pass

    return app

运行应用

  1. export FLASK_APP=flaskrexport FLASK_ENV=development ,flask run

定义和操作数据库

连接数据库

  1. 当使用SQL数据时,首先要创建一个数据库的连接。所有查询和操作都要通过该连接来执行,执行完毕后关闭该连接。
  2. 使用g对象存储数据库连接,可以在一个请求中多次使用

创建表

  1. 创建sql文件储存创建空表的SQL命令,在 db.py 文件中添加init_db函数,用于运行这个 SQL 命令
  2. open_resource() 打开一个文件,该文件名路径是相对于 flaskr 包的,click.command() 定义一个名为 init-db 命令行

在应用中注册

  1. 使用工厂函数后函数调用前实际上不存在应用实例,因此close_db,init_db等函数无法使用,因此需要定义一个注册函数把应用作为参数完成创建和关闭数据库函数的注册
def init_app(app):
    app.teardown_appcontext(close_db)
    app.cli.add_command(init_db_command)
  1. 注册函数需要导入添加到应用工厂函数中

初始化数据库文件

  1. 运行init-db命令,在实例文件夹中创建数据库文件(该文件夹不在项目文件夹中,也不会git上传)

蓝图和视图

创建蓝图

  1. 蓝图方式是把视图函数注册到蓝图,然后在工厂函数中把蓝图注册到应用,每个功能模块可以创建一个单独的蓝图
  2. 创建方式:bp = Blueprint('auth', __name__, url_prefix='/auth')前缀是可选的
  3. 注册蓝图:先导入模块,再使用app.register_blueprint(blueprint)函数

认证蓝图的注册视图

  1. 使用blueprint.route注册URL
  2. 如果用户提交表单,开始表单验证
    • 从request中提取表单输入值,定义空error值
    • 验证数据不为空,如果为空定义error信息用于后面flash()
    • 查询数据库验证是否已被注册(使用占位符而非直接代入参数以防范SQL注入攻击),如果为空定义error信息
    • 无error则验证成功,在数据库中插入新用户数据,密码存储前进行哈希处理generate_password_hash()
    • 向数据库提交数据后重定向到登录页面,如果验证失败,那么会向用户显示一个出错信息
  3. 如果用户注册出错或使用GET方法访问视图,渲染一个注册表单页面
@bp.route('/register', methods=('GET', 'POST'))
def register():
    if request.method == 'POST':
        username = request.form['username']
        password = request.form['password']
        db = get_db()
        error = None

        if not username:
            error = 'Username is required.'
        elif not password:
            error = 'Password is required.'
        elif db.execute(
            'SELECT id FROM user WHERE username = ?', (username,)
        ).fetchone() is not None:
            error = 'User {} is already registered.'.format(username)

        if error is None:
            db.execute(
                'INSERT INTO user (username, password) VALUES (?, ?)',
                (username, generate_password_hash(password))
            )
            db.commit()
            return redirect(url_for('auth.login'))

        flash(error)

    return render_template('auth/register.html')

登录视图

  1. 逻辑与注册视图基本相同
  2. 首先在数据库中查询用户并存放于变量中
  3. 验证用户是否存在及密码是否正确,check_password_hash()哈希提交的密码并安全的比较哈希值
  4. 如果验证通过(error为None),将用户id储存于session中(先清除),并重定向到主页。flask会自动将session添加进cookie中
  5. 注册before_app_request视图检验是否已登录:判断session中是否存在user_id,如果有则向数据库查询id并将返回的信息存储在g.user中
@bp.before_app_request  # 在处理每个请求前都会调用该函数
def load_logged_in_user():
    user_id = session.get('user_id')

    if user_id is None:
        g.user = None
    else:
        g.user = get_db().execute(
            'SELECT * FROM user WHERE id = ?', (user_id,)
        ).fetchone()

注销视图

  1. 处理逻辑:清除session然后重定向到主页
@bp.route('/logout')
def logout():
    session.clear()
    return redirect(url_for('index'))

在其他视图中需要认证

  1. 实现一个认证装饰器,只有已经登录的用户可以执行被装饰的视图(未登录则被重定向至登录页面)
def login_required(view):
    # 将原函数对象的指定属性复制给包装函数对象,默认有module、name、doc,或者通过参数选择,确保装饰器不会对被装饰函数造成影响
    @functools.wraps(view)
    def wrapped_view(**kwargs):
        if g.user is None:
            return redirect(url_for('auth.login'))

        return view(**kwargs)

    return wrapped_view

模板

  1. Jinja引擎会自动转义HTML模板中的任何数据。即直接渲染用户的输入是安全的
  2. {{ 和 }} 这间的东西是一个会输出到最终文档的静态表达式。{%和%}之间的东西表示流程控制语句,如if和for

基础布局

  1. 创建一个base html,而不是每个页面都重写。base中定义多个block,在继承的模板中被重载
  2. g 对象在模板中是自动可用的,base中检测g.user是否存在(before_app_request装饰器检测是否已登录),分别显示用户名、注销连接或注册、登录连接
  3. 模板中尽量使用url_for()转换URL而不是直接写出
  4. 在页面的主要内容,先用for in表达式展示通过get_flashed_messages()返回视图中的flash信息
  5. 最后定义一个block content展示页面主要内容。
  6. base模板中可定义的块:{% block title %} {% block header %} {% block content %}

注册模板

  1. 可以直接{% block title %} 放在 {% block header %} 内部,两者都会在模板中生效
{% extends 'base.html' %}

{% block header %}
  <h1>{% block title %}Register{% endblock %}</h1>
{% endblock %}

{% block content %}
  <form method="post">
    <label for="username">Username</label>
    <input name="username" id="username" required>
    <label for="password">Password</label>
    <input type="password" name="password" id="password" required>
    <input type="submit" value="Register">
  </form>
{% endblock %}

静态文件

  1. Flask自动生成一个static视图,用来生成URL引用static文件夹中的css等静态文件:url_for('static', filename='...')

博客蓝图

  1. 博客蓝图需要实现的功能:展示所有posts,允许已登录用户创建posts,允许posts的作者对其进行编辑或删除

创建蓝图

  1. 定义蓝图
from flask import (
    Blueprint, flash, g, redirect, render_template, request, url_for
)
from werkzeug.exceptions import abort

from flaskr.auth import login_required
from flaskr.db import get_db

bp = Blueprint('blog', __name__)
  1. 注册到应用工厂,使用app.add_url_rule()关联'/'URL到蓝图中的index端点(这样其他视图中指向'/'的链接都会关联到蓝图中的index视图)
def create_app():
    app = ...
    # existing code omitted

    from . import blog
    app.register_blueprint(blog.bp)
    app.add_url_rule('/', endpoint='index')

    return app

Index视图

  1. 视图函数
@bp.route('/')
def index():
    db = get_db()
    posts = db.execute(
        'SELECT p.id, title, body, created, author_id, username'
        ' FROM post p JOIN user u ON p.author_id = u.id'
        ' ORDER BY created DESC'
    ).fetchall()
    return render_template('blog/index.html', posts=posts)
  1. 模板文件
{% extends 'base.html' %}

{% block header %}
  <h1>{% block title %}Posts{% endblock %}</h1>
  {% if g.user %}
    <a class="action" href="{{ url_for('blog.create') }}">New</a>
  {% endif %}
{% endblock %}

{% block content %}
  {% for post in posts %}
    <article class="post">
      <header>
        <div>
          <h1>{{ post['title'] }}</h1>
          <div class="about">by {{ post['username'] }} on {{ post['created'].strftime('%Y-%m-%d') }}</div>
        </div>
        {% if g.user['id'] == post['author_id'] %}
          <a class="action" href="{{ url_for('blog.update', id=post['id']) }}">Edit</a>
        {% endif %}
      </header>
      <p class="body">{{ post['body'] }}</p>
    </article>
    {% if not loop.last %}
      <hr>
    {% endif %}
  {% endfor %}
{% endblock %}

Create视图

  1. 视图函数(逻辑与注册视图基本相同)
@bp.route('/create', methods=('GET', 'POST'))
@login_required
def create():
    if request.method == 'POST':
        title = request.form['title']
        body = request.form['body']
        error = None

        if not title:
            error = 'Title is required.'

        if error is not None:
            flash(error)
        else:
            db = get_db()
            db.execute(
                'INSERT INTO post (title, body, author_id)'
                ' VALUES (?, ?, ?)',
                (title, body, g.user['id'])
            )
            db.commit()
            return redirect(url_for('blog.index'))

    return render_template('blog/create.html')
  1. 模板文件
{% extends 'base.html' %}

{% block header %}
  <h1>{% block title %}New Post{% endblock %}</h1>
{% endblock %}

{% block content %}
  <form method="post">
    <label for="title">Title</label>
    <input name="title" id="title" value="{{ request.form['title'] }}" required>
    <label for="body">Body</label>
    <textarea name="body" id="body">{{ request.form['body'] }}</textarea>
    <input type="submit" value="Save">
  </form>
{% endblock %}

Update视图

  1. 定义get_post(id, check_author=True)函数获取指定id的post内容
def get_post(id, check_author=True):
    post = get_db().execute(
        'SELECT p.id, title, body, created, author_id, username'
        ' FROM post p JOIN user u ON p.author_id = u.id'
        ' WHERE p.id = ?',
        (id,)
    ).fetchone()

    if post is None:
        abort(404, "Post id {0} doesn't exist.".format(id))

    if check_author and post['author_id'] != g.user['id']:
        abort(403)

    return post
  1. 视图函数,\中的id参数由模板中的url_for('blog.update', id=post['id'])传入
@bp.route('/<int:id>/update', methods=('GET', 'POST'))
@login_required
def update(id):
    post = get_post(id)

    if request.method == 'POST':
        title = request.form['title']
        body = request.form['body']
        error = None

        if not title:
            error = 'Title is required.'

        if error is not None:
            flash(error)
        else:
            db = get_db()
            db.execute(
                'UPDATE post SET title = ?, body = ?'
                ' WHERE id = ?',
                (title, body, id)
            )
            db.commit()
            return redirect(url_for('blog.index'))

    return render_template('blog/update.html', post=post)
  1. 模板文件(可重构后与Create合并为同一html文件)
{% extends 'base.html' %}

{% block header %}
  <h1>{% block title %}Edit "{{ post['title'] }}"{% endblock %}</h1>
{% endblock %}

{% block content %}
  <form method="post">
    <label for="title">Title</label>
    <input name="title" id="title"
      value="{{ request.form['title'] or post['title'] }}" required>
    <label for="body">Body</label>
    <textarea name="body" id="body">{{ request.form['body'] or post['body'] }}</textarea>
    <input type="submit" value="Save">
  </form>
  <hr>
  <form action="{{ url_for('blog.delete', id=post['id']) }}" method="post">
    <input class="danger" type="submit" value="Delete" onclick="return confirm('Are you sure?');">
  </form>
{% endblock %}
  1. 参数 {{ request.form['title'] or post['title'] }} 用于按顺序选择在表单显示什么 数据。当表单还未提交时,显示原 post 数据。但是,如果提交了非法数据,然后 需要显示这些非法数据以便于用户修改时,就显示 request.form 中的数据。 request 是又一个自动在模板中可用的变量。

Delete视图

  1. 只有视图函数,无需模板文件。删除的处罚按钮位于update页面中
@bp.route('/<int:id>/delete', methods=('POST',))
@login_required
def delete(id):
    get_post(id)
    db = get_db()
    db.execute('DELETE FROM post WHERE id = ?', (id,))
    db.commit()
    return redirect(url_for('blog.index'))

项目可安装化

  1. 构建一个可分发文件,使任何环境都可以像安装flask一样使用标准Python工具部署项目包(包括所有的依赖包),同时可以分离测试和开发环境
  2. 创建setup.py文件描述项目及从属文件
from setuptools import find_packages, setup

setup(
    name='flaskr',
    version='1.0.0',
    packages=find_packages(),
    include_package_data=True,
    zip_safe=False,
    install_requires=[
        'flask',
    ],
)
  1. 创建MANIFET.in文件,赋值从属的静态文件但排除其中的所有字节文件
include flaskr/schema.sql
graft flaskr/static
graft flaskr/templates
global-exclude *.pyc
  1. 安装项目:使用 pip 在虚拟环境中安装项目pip install -e

测试覆盖

  1. 应该对代码中的每个函数、每个条件分支进行单元测试,以防出现不可预期的错误
  2. 进行测试需要用到的三方库:pytest(为出错的assert语句展示详细信息)及coverage(分析被实际执行(如被测试)的代码数量)
  3. 创建数据库用于单独存储测试数据并写入测试数据(写个SQL文件)
  4. @pytest.fixture,用来包装一个函数(比如登录注销等配置过程),在调用测试用例时将fixture包装的函数名作为参数调用,会在测试用例执行之前调用被包装的函数。用来省略每次测试都需要重新建数据库、创建应用实例、登录等过程
import os
import tempfile

import pytest
from flaskr import create_app
from flaskr.db import get_db, init_db

with open(os.path.join(os.path.dirname(__file__), 'data.sql'), 'rb') as f:
    _data_sql = f.read().decode('utf8')


@pytest.fixture
def app():
    db_fd, db_path = tempfile.mkstemp()

    app = create_app({
        'TESTING': True,
        'DATABASE': db_path,
    })

    with app.app_context():
        init_db()
        get_db().executescript(_data_sql)

    yield app

    os.close(db_fd)
    os.unlink(db_path)


@pytest.fixture
def client(app):
    return app.test_client()


@pytest.fixture
def runner(app):
    return app.test_cli_runner()
  1. 后续具体测试省略...

部署产品

  1. 安装wheel库,构建发行文件:python setup.py bdist_wheel,使用pip install安装(项目及相关依赖)
  2. 配置随机密钥:python -c 'import os; print(os.urandom(16))'
  3. Flask 的内建服务器不适用于生产,生产环境应使用生产级别的服务器

注:转载本文,请与作者联系




如果觉得文章对您有价值,请作者喝杯咖啡吧

|
donate qrcode

欢迎通过微信与我联系

wechat qrcode

0 Comments latest

No comments.