Cheat Sheat

构建路由（注册函数）

  • app.route('/')

创建 URL

  • url_for

处理模板

  • render_template

文件夹

  • static

    • 静态文件

      • 存放 css、javascript 文件等

      • url 生成

        1

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

  • templates

    • 模板文件

      • 存放 html 模板

      • 处理模板

        1

        render_template('hello.html', name=name)

配置

app.config

sqlalchemy

https://flask-sqlalchemy.palletsprojects.com/en/2.x/quickstart/

表 db.Model

单个表

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15

from flask import Flask
from flask_sqlalchemy import SQLAlchemy

app = Flask(__name__)
app.config['SQLALCHEMY_DATABASE_URI'] = 'sqlite:////tmp/test.db'
db = SQLAlchemy(app)


class User(db.Model):
    id = db.Column(db.Integer, primary_key=True)
    username = db.Column(db.String(80), unique=True, nullable=False)
    email = db.Column(db.String(120), unique=True, nullable=False)

    def __repr__(self):
        return '<User %r>' % self.username

多个表

https://www.cnblogs.com/liangmingshen/p/9769975.html

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16

class User(Base):
    __tablename__ = 'user'
    id = Column(Integer, primary_key=True)
    name = Column(String)

    # * Address 关联的表
    addresses = relationship("Address", backref="user")


class Address(Base):
    __tablename__ = 'address'
    id = Column(Integer, primary_key=True)
    email = Column(String)

    # * 外键
    user_id = Column(Integer, ForeignKey('user.id'))

  1. 关联方法

    • 当前表：db.relationship 声明关联的表
    • 关联的表：外键 ForeignKey

  2. 当前表 链接 关联表

    • user.address

  3. 关联表 链接 当前表

    • address.user

  4. 注意：

    • 这里必须使用 backref 了

      1
2
3
4

      addresses = db.relationship("Address", backref="user")

# * 高级用法
addresses = db.relationship("Address", backref=db.backref("users", lazy=True))
backref

https://www.cnblogs.com/liangmingshen/p/9769975.html

backref 用于在关系另一端的类中快捷地创建一个指向当前类对象的属性, 而当需要对那个属性指定参数时使用 db.backref()。

操作

创建记录

  • admin = User(username='admin', email='admin@example.com')

修改与提交

在表上修改
  • address.users.append(lucy)
在数据库上修改
  • db.session.add(user)
  • db.session.commit()

查询

  • User.query.all()

    • 查询整个表

  • User.query.filter_by(username='admin').first()

    • 查询过滤

加载问题

  • 对于 lazy-loading

    • 需要在必要时加载关联的 backref 表

      1
2
3
4
5

      >>> from sqlalchemy.orm import joinedload
>>> query = Category.query.options(joinedload('posts'))
>>> for category in query:
...     print category, category.posts
<Category u'Python'> [<Post u'Hello Python!'>, <Post u'Snakes'>]

路由 Routing

https://flask.palletsprojects.com/en/1.1.x/quickstart/#routing

https://flask.palletsprojects.com/en/1.1.x/api/#url-route-registrations 官方文档

https://www.cnblogs.com/amyleell/articles/9178596.html 路由的工作原理

  • 经典路由 canonical URL

    • 经典路由都是一 “/” 结尾

@app.route()装饰器 与 app.add_url_rule() 方法

  • 可以相互替代

  • 共有参数

    • rule

      • URL 规则，路由路径

    • view_func

      • 视图函数名称，用来处理路径的函数

    • defaults=None

      • 默认值, url 的缺省默认值，当 URL 中无参数，函数需要参数时，使用 defaults={'k':'v'}为函数提供参数

    • endpoint=None

      • 名称，用于反向生成 URL，即： url_for('名称')

    • methods=None

      • 允许的请求方式，如：["GET","POST"]

    • strict_slashes=None

      • 对 URL 最后的 / 符号是否严格要求
    • 有没有最有的‘/’，是否有区别
    • redirect_to=None
    • subdomain=None
    1
2
3
4

    def index():
    return "Index"

self.add_url_rule(rule='/index.html', endpoint="index", view_func=index, methods=["GET","POST"])
    1
2

    app.add_url_rule(rule='/index.html', endpoint="index", view_func=index, methods=["GET","POST"])
app.view_functions['index'] = index
     1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12

    def auth(func):
    def inner(*args, **kwargs):
        print('before')
        result = func(*args, **kwargs)
        print('after')
        return result
    return inner

@app.route('/index.html', methods=['GET', 'POST'], endpoint='index')
@auth
def index():
    return 'Index'

@app.route() 的 被修饰函数 与 app.add_url_rule 的 view_functions 一致

端点 endpoint 与 视图函数 view_function

https://www.cnblogs.com/amyleell/articles/9178596.html

  • 路由的处理顺序

    • Url –> endpoint –> view_function
  • 一般情况 endpoint 与 view_function 一致

末尾 '/' 的处理

  • 程序的 rule='xxx', 带有 '/'

    • 用户输入路径

      • 末尾不带 '/'

    • 程序自动转换成带有 '/' 的 rule

      • 末尾带 '/'
    • 正常处理

  • 程序的 rule='xxx', 不带 ''

    • 用户输入路径

      • 末尾不带 '/'

    • 正常处理

      • 末尾带 '/'
    • 发生 404 错误

同一个 view_func, 多个 rule

1
2
3
4

@app.route('/users/', defaults={'page': 1})
@app.route('/users/page/<int:page>')
def show_users(page):
    pass

class View

https://flask.palletsprojects.com/en/1.1.x/views/

  • 属性和方法有规定名称和样式

  • 使用方法

    • 使用 class method

      • your_view.as_view('your_view')

flask.views.View 基类

flask.views.MethodView

Request Context

  • 针对一个 request context

    • 即环境，包含各种变量，这些对于一个 request 都是全局的
    • 但是针对不同的 Request 是隔离的

手动测试 request context 的方法

https://flask.palletsprojects.com/en/1.1.x/reqcontext/

  • app.test_request_context()

    1
2
3
4
5
6
7

    def generate_report(year):
    format = request.args.get('format')
    ...

with app.test_request_context(
        '/make_report/2017', data={'format': 'short'}):
    generate_report()

一个 request 的处理过程，相关的函数

https://flask.palletsprojects.com/en/1.1.x/reqcontext/#callbacks-and-errors

  • 按顺序排列

    1. before_request()

      • 这是多个函数
      • 但是只会最多触发一个
      • 如果触发，并返回一个 Response, 则跳过 view_func

    2. view_func

      • 视图函数
      • 处理 request, 并返回一个 Response

    3. after_request()

      • 多个函数
      • 修改前面生成的 Response, 或者制作一个全新的 Response

    4. errorhandler()

      • 如果在 after_request() 之中，或之前发生异常，会被调用
      • 产生一个异常处理的 Response
      • 如果没有手动设置，则生成 500 Internal Server Error Response.

    5. teardown_request() 和 teardown_appcontext()

      • 都是多个函数
      • 即使发生异常也会被调用

回调函数

teardown_request

栈弹出 RequestContext, 回调函数

  • @app.teardown_request
 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25

  from flask import Flask, request

  app = Flask(__name__)

  @app.route('/')
  def hello():
print('during view')
return 'Hello, World!'

  @app.teardown_request
  def show_teardown(exception):
print('after with block')

  with app.test_request_context():
print('during with block')

  # teardown functions are called after the context with block exits

  with app.test_client() as client:
client.get('/')
# the contexts are not popped even though the request ended
print(request.path)

  # the contexts are popped and teardown functions are called after
  # the client with block exits

errorhandler

异常回调函数

  • @app.errorhandler(error_code)

    • eg: @app.errorhandler(404)
    1
2
3
4
5

    from flask import render_template

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

after_request

@after_this_request

用来修饰 view_func 生成的 Response https://flask.palletsprojects.com/en/1.1.x/api/#flask.after_this_request

1
2
3
4
5
6
7

@app.route('/')
def index():
    @after_this_request
    def add_header(response):
        response.headers['X-Foo'] = 'Parachute'
        return response
return 'Hello World!'

手动测试 flask

app.test_client()

1
2
3
4

with app.test_client() as client:
    client.get('/')
    # the contexts are not popped even though the request ended
    print(request.path)

app.test_request_context

1
2
3
4
5
6
7

from flask import request

with app.test_request_context('/hello', method='POST'):
    # now you can do something with the request until the
    # end of the with block, such as basic assertions:
    assert request.path == '/hello'
    assert request.method == 'POST'

flask Request

https://flask.palletsprojects.com/en/1.1.x/quickstart/#accessing-request-data

1

from flask import request

直接访问 flask.requset 对象的属性

  • request.method == 'POST'

request methods

  • request.method

requset 参数

  • request.args

    • 一个 dict 对象

request 上传的文件 files

  • request.files

flask Response

https://flask.palletsprojects.com/en/1.1.x/quickstart/#about-responses

返回值，转换成 Response 对象

jsonify 与 json

把 dict 转换成 Response
  • 使用 jsonify(your_dict), 把 your_dict, 转换成 Response
  • 这种转换，是 flask 自动完成的，当返回值是 dict 类型时，不需要具体 手动调用 jsonify
也可用于非 dict 类型
1
2
3
4

@app.route("/users")
def users_api():
    users = get_all_users()
    return jsonify([user.to_json() for user in users])

转换规则

  1. 本身就是 Response 对象

    • 不作处理

  2. string, 字符串

    • 使用默认参数，转换成 Response

  3. dict

    • 使用 jsonify

  4. tuple

    • 可以额外提供更多信息
    • 但是要有一定的格式

    • 格式

      • (response, status)
      • (response, headers)
      • (response, status, headers)

    • 解说

      • status
      • status_code
      • 将会替换掉 Response 中的 status_code

      • headers

        • list or dict

  5. 其它

    • If none of that works, Flask will assume the return value is a valid WSGI application and convert that into a response object.
    • 当作 WSGI application, 并转换成 Response

自己制作 Response object

模板

flask.render_template(filename, **context)
1

flask.render_template(template_name_or_list, **context)

  • template: 第一个参数

    • html 模板
    • 注意：
  • 这里是 template 文件夹下的文件
  • 如果是 list, 使用第一个有效的

  • **context: 其余参数

    • 以关键字参数的形式，指定模板中出现的变量
flask.render_template_string(template_string, **context)

生成 Response

个性化配置 Response

flask.make_response(*args)

  • 用法样式

    • make_response()

  • 生成空 Response()

    • make_response(content)
    • make_response(content, status_code)
1
2
3
4
5

@app.errorhandler(404)
def not_found(error):
    resp = make_response(render_template('error.html'), 404)
    resp.headers['X-Something'] = 'A value'
    return resp

flask Sessions 会话

用于跨 Request 数据共享，对于同一个用户

  • 基于 cookies
  • 和 signs the cookies cryptographically(cookies 加密) https://flask.palletsprojects.com/en/1.1.x/quickstart/#sessions https://flask.palletsprojects.com/en/1.1.x/api/#sessions

  • 权限问题

    • 用户没有 secret key 密钥，只能查询，不能修改 cookie
    • 使用 sessions, 必须设置 secret key

  • 使用

    1. 设置 app.secret_key

    2. 把 session 当作 dict 使用

      • session['username']
1
2
3
4
5
6
7
8

app.secret_key = b'_5#y2L"F4Q8z\n\xec]/'

session['username'] = request.form['username']

if 'username' in session:
    return 'Logged in as %s' % escape(session['username'])

session.pop('username', None)

密钥制作工具

1
2

$ python -c 'import os; print(os.urandom(16))'
b'_5#y2L"F4Q8z\n\xec]/'

html 转义

把一些特殊字符（&, <, >, ', "）等，转换成 html 能识别的字符

flask redirect 跳转

https://flask.palletsprojects.com/en/1.1.x/quickstart/#redirects-and-errors

 1
 2
 3
 4
 5
 6
 7
 8
 9
10

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()

flask.redirect(location, code=302, Response=None)

返回 Response, 把客户端引导到 location

flask.url_for(endpoint, **values)

制作一个 url

flask Blueprint

用来组织 flask views

  • 原来

    • app –> views

  • 现在

    • app –> Blueprint –> views

  • 特征

    • 有些类似 app, 但又不是 app

    • 适合大型项目

      • 模块化
      • 结构化
    • 可以用 Blueprint 提供 template filters, static files, templates and other utilities
    • 一个 Blueprint 可以没有 view funcions
    • 共享 app 的配置

  • 绑定位置

    • 一个 URL prefix 或 '/'
    • 子域名 subdomain

    • 特别的

      • 可以在不同的 URL rule 上，重复绑定

定义 Blueprint

  • 定义：

    • your_blu = Blueprint(name: str, import_name: str, static_folder_name: str)

  • route 使用

    • @your_blu.route()

:定义 Blueprint:

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13

  from flask import Blueprint, render_template, abort
  from jinja2 import TemplateNotFound

  simple_page = Blueprint('simple_page', __name__,
     template_folder='templates')

  @simple_page.route('/', defaults={'page': 'index'})
  @simple_page.route('/<page>')
  def show(page):
try:
   return render_template('pages/%s.html' % page)
except TemplateNotFound:
   abort(404)

:END:

绑定到 app

  • app.register_blueprint(your_blu)

    • 绑定到 '/your_blu.name'

  • app.register_blueprint(simple_page, url_prefix='/given_pages')

    • 绑定到 '/given_pages'
1
2
3
4
5

from flask import Flask
from yourapplication.simple_page import simple_page

app = Flask(__name__)
app.register_blueprint(simple_page)

资源文件夹

  • 一般是 "项目目录/blueprint_name"

Resource folder 资源文件夹

  • your_blu.open_resource()
1
2

with simple_page.open_resource('static/style.css') as f:
    code = f.read()

Static Files

  • Blueprint 拥有 url_prefix='/given_pages'

    • 文件夹： /your_blu.url_prefix/static_folder_name

      1
2

      admin = Blueprint('admin', __name__, static_folder='static')
# static folder --> /admin/static

    • 没有 url_prefix

      • 获取不到 static files 文件夹
    • 因为 /static 属于整个 app, 优先级更高（冲突）

Templates

1

admin = Blueprint('admin', __name__, template_folder='templates')

Building URLs

  • 使用 url_for

  • 其它 Blueprint

    • 以 "." 分割 Blueprint 名称 和 end_point

      1

      url_for('admin.index')

  • 当前 Blueprint

    • 以 "." 开头 + end_point

      1

      url_for('admin.index')

flask flash

flask(message: str) 把消息存储到 session 中，在模板中使用 get_flashed_message() 获取生成的列表

  • 工作原理

    1. flask 把数据存储到 session 中

    2. view_func 返回一个模板 或 跳转到一个返回模板的 view_func

      • eg: a) return Your_temp b) return redirect(url_for('hello'))

    3. 在模板中取出，之前存储的数据

      • 使用 get_flashed_message()提取

https://www.cnblogs.com/xiaxiaoxu/p/10468103.html

flask instance

instance folder

https://flask.palletsprojects.com/en/0.12.x/config/#instance-folders

  • 作用

    • 用来规避 VCS 软件的版本控制
    • 放置配置文件

  • 可以自己指定 路径， 也可以是默认值

    • 注意：自己指定，必须是绝对路径

      1

      app = Flask(__name__, instance_path='/path/to/instance/folder')

  • 默认值

    • 项目根目录

      /myapp.py /instance

Flask app 与 instance_relative_config

是否配置文件查询路径在 instance 文件夹

1
2
3

app = Flask(__name__, instance_relative_config=True)
app.config.from_object('yourapplication.default_settings')
app.config.from_pyfile('application.cfg', silent=True)

Flask.open_instance_resource()

打开 instance 文件夹下文件的快捷方式

1
2
3
4
5
6
7

filename = os.path.join(app.instance_path, 'application.cfg')
with open(filename) as f:
    config = f.read()

# or via open_instance_resource:
with app.open_instance_resource('application.cfg') as f:
    config = f.read()

flask sqlite3

https://flask.palletsprojects.com/en/1.1.x/tutorial/database/

连接

  • flask.g.db = sqlite3.connect

相关变量

flask.g

https://flask.palletsprojects.com/en/1.1.x/api/#flask.g 跨 Application Context 的变量

  • 用来存储每个 request 共有的变量

关闭

  • db.close()
 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24

import sqlite3

import click
from flask import current_app, g
from flask.cli import with_appcontext


def get_db():
    if 'db' not in g:
    g.db = sqlite3.connect(
        current_app.config['DATABASE'],
        detect_types=sqlite3.PARSE_DECLTYPES
    )
    # * sqlite3.Row factory
    g.db.row_factory = sqlite3.Row

    return g.db


def close_db(e=None):
    db = g.pop('db', None)

    if db is not None:
    db.close()

执行 ".sql" 文件

db.executescript(your_file_read_as_string)

1
2

with current_app.open_resource('schema.sql') as f:
    db.executescript(f.read().decode('utf8'))

url_for

Flask logging

注意

  • 关于 root_logger 问题

    • 即，除掉在 app.logger 以外 模块的 logging 输出配置

    • 解决方法

      • 在创建 app 之前，先配置完成 logging

      • eg:

         1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29

        from logging.config import dictConfig

dictConfig({
    'version': 1,
    'formatters': {'default': {
    'format': '[%(asctime)s] %(levelname)s in %(module)s: %(message)s',
    }},
    'handlers': {
    'rotate_file': {
        'class': 'logging.handlers.RotatingFileHandler',
        'filename': 'record.log',
        'mode':'a',
        'maxBytes': 1024*1024*10,
        'backupCount': 10,
        'encoding': 'utf-8',
        'formatter': 'default'
    },
    'wsgi': {
        'class': 'logging.StreamHandler',
        'stream': 'ext://flask.logging.wsgi_errors_stream',
        'formatter': 'default'
    }},
    'root': {
    'level': 'INFO',
    'handlers': ['wsgi', 'rotate_file']
    }
})

app = Flask(__name__)

监控插件

flask_monitoringdashboard

async + flask

参考：

注意：

  • python 3.8 有 asyncio bug, 不推荐使用

布署 deploy

参考：

windows 支持

  • waitress

文件下载（向客户端发送文件） download file

方法：

  1. 通过 flask.send_file() 函数直接转递文件路径

    • 它会通过参数 file_path_or_io_object, 或者 download_name, 再或者 mime_type 来或者最终的 mime_type
    • 如果没有找到 mime_type, 会选用 binary 格式发送文件，即 "application/octet-stream"

    • 参数 as_attachment 会设置 http header Content-Disposition: attachment

      • 通知 Web 端，接收的是文件，不要直接展示
    • 返回 Response 类型

    • 特点：

      • 下载时，会显示总大小，和进度条

    • 例子

      1
2
3
4
5
6

      @app.route("/file/whole/send_file", methods=["GET"])
def send_whole():
    return send_file(
        "/mnt/e/Downloads/torch-2.1.1+cu118-cp310-cp310-linux_x86_64.whl",
        download_name="hello.whl",
    )

  2. 通过流 stream 实现流式下载

    1. 特点：

      • 下载时，没有百分比和移动的进度条，没有总大小
    2. 通过 generator 实现
    3. 通过 generator + flask.stream_with_context 实现

    4. flask.send_file() + 函数实现

      • send_file 接收一个返回 IO 对象的函数

    5. flask.send_file + IO 对象

      • eg:

        1

        return send_file(open("/path/to/large_file", "rb"), download_name="example.data")

更多方法，参考：

stream 流式文件下载

自定义设置：

  • http header: Content-Disposition: attachment; filename="example.pdf"

    • 通知客户端，下载数据是附件，不要直接展示
    • filename="example.pdf", 给出文件的名称

方法：

  1. 直接使用 generator:

    • 这是一种基础用法

    • generator 返回的文本，当作文件内容

      1
2
3
4
5
6

      @app.route('/large.csv')
def generate_large_csv():
    def generate():
        for row in iter_all_rows():
            yield f"{','.join(row)}\n"
    return generate(), {"Content-Type": "text/csv"}

  2. 使用 flask.stream_with_context

    • 特点：

      • 允许在 generator 里面使用 flask.request 获取请求中的数据
    • flask.stream_with_context 是一个装饰器