【Flask】04-高階特性

阿新 • • 發佈：2020-07-18

工欲善其事，必先利其器。

檔案上傳

用 Flask 處理檔案上傳很容易，只要確保不要忘記在你的 HTML 表單中設定 enctype="multipart/form-data" 屬性就可以了。否則瀏覽器將不會傳送你的檔案。

已上傳的檔案被儲存在記憶體或檔案系統的臨時位置。你可以通過請求物件 files 屬性來訪問上傳的檔案，save()方法用於把上傳檔案儲存到伺服器的檔案系統中。

檔案上傳的基本原理實際上很簡單，基本上是：

一個帶有 enctype=multipart/form-data 的 標記，標記中含有一個 。
應用通過請求物件的 files 字典來訪問檔案。
使用檔案的 save()

方法把檔案永久地儲存在檔案系統中。

案例：上傳檔案到一個指定目錄

前端程式碼

<!DOCTYPE html>
<html lang="en">
<head>
    <meta charset="UTF-8">
    <title>Title</title>
</head>
<body>
<form action="http://127.0.0.1:5000/upload/" method="post" enctype="multipart/form-data">
    <input type="file" name="file">
    <input type="submit" value="上傳">
</form>
</body>
</html>

後端程式碼

import os
from flask import Flask, flash, request, redirect, url_for, render_template
from werkzeug.utils import secure_filename

UPLOAD_FOLDER = os.path.dirname(os.path.abspath(__name__))
ALLOWED_EXTENSIONS = {'txt', 'pdf', 'png', 'jpg', 'jpeg', 'gif'}

app = Flask(__name__)
app.config['UPLOAD_FOLDER'] = UPLOAD_FOLDER
app.config['SECRET_KEY'] = "ahfkkalahp9qnofnovna"
app.config['MAX_CONTENT_LENGTH'] = 16 * 1024 * 1024 # 尺寸限制為 16 M


def allowed_file(filename):
    # 判斷後綴
    return '.' in filename and \
           filename.rsplit('.', 1)[1].lower() in ALLOWED_EXTENSIONS


@app.route("/upload/", methods=["POST", "GET"])
def upload():
    if request.method == "GET":
        return render_template("upload.html")
    else:
        if 'file' not in request.files:　# 判斷表單name
            flash('No file part')
            return redirect(request.url)
        file = request.files['file']
        if file.filename == '': # 判斷檔名稱
            flash('No selected file')
            return redirect(request.url)
        if file and allowed_file(file.filename):　# 判斷檔案字尾是否符合ALLOWED_EXTENSIONS
            filename = secure_filename(file.filename)
            file.save(os.path.join(app.config['UPLOAD_FOLDER'], filename))
            return "upload ok"

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

UPLOAD_FOLDER ：是上傳文件要儲存的目錄。
ALLOWED_EXTENSIONS ：是允許上傳的副檔名的集合。
MAX_CONTENT_LENGTH：限制檔案大小
SECRET_KEY：post請求需要配置該引數
secure_filename() ：假設filename = "../../../../home/username/.bashrc"，你會把它和 UPLOAD_FOLDER 結合在一起，那麼使用者就可能有能力修改一個伺服器上的檔案，這個檔案本來是使用者無權修改的

現在來看看函式是如何工作的：

>>> secure_filename('../../../../home/username/.bashrc')
'home_username_.bashrc'

上傳擴充套件　Flask-Uploads

會話

Cookies

要訪問 cookies ，可以使用 cookies屬性。可以使用響應物件的 set_cookie方法來設定 cookies 。請求物件的 cookies屬性是一個包含了客戶端傳輸的所有 cookies 的字典。

讀取 cookies

from flask import request

@app.route('/')
def index():
    username = request.cookies.get('username')

儲存 cookies

from flask import make_response

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

注意， cookies 設定在響應物件上。通常只是從檢視函式返回字串， Flask 會把它們轉換為響應物件。如果你想顯式地轉換，那麼可以使用 make_response() 函式，然後再修改它。

set_cookie(key，value =''，max_age = None，expires = None，path ='/'，domain = None，secure = False，httponly = False，samesite = None)

引數	介紹
`key`	鍵
`value`	值
`max_age`	過期時間，單位秒
`expries`	過期時間，`datetime`型別，這個時間需要設定為格林尼治時間，也就是要距離北京少8個小時的時間。
`path`	表示儲存瀏覽器根目錄
`domain`	設定跨域
`secure`	為`True`表示僅支援`HTTPS`
`httponly`	禁止`JavaScript`訪問`Cookie`。這是個`Cookie`標準的擴充套件，可能不是所有瀏覽器都支援。
`samesite`	限制`cookie`的範圍，只能訪問當前設定的站點

刪除 cookies

   # 原始碼
    def delete_cookie(self, key, path="/", domain=None):
        """Delete a cookie.  Fails silently if key doesn't exist.

        :param key: the key (name) of the cookie to be deleted.
        :param path: if the cookie that should be deleted was limited to a
                     path, the path has to be defined here.
        :param domain: if the cookie that should be deleted was limited to a
                       domain, that domain has to be defined here.
        """
        self.set_cookie(key, expires=0, max_age=0, path=path, domain=domain)

Session

除了請求物件之外還有一種稱為 session 的物件，允許你在不同請求之間儲存資訊。這個物件相當於用金鑰簽名加密的 cookie ，即使用者可以檢視你的 cookie ，但是如果沒有金鑰就無法修改它。

案例：

import os
from flask import session, Flask

app = Flask(__name__)
app.config['SECRET_KEY'] = os.urandom(16)

@app.route("/set-session/")
def set_session():
    session["username"] = "jack"
    return "set session ok"

@app.route("/get-session/")
def get_session():
    username = session.get("username")
    if not username:
        return "get session fail"
    else:
        return username

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

設定 Session

session[key] = value

獲取 Session

session[key]
session.get(key)

刪除 Session某個鍵值

session.pop(key)
del session[key]
session.clear() # 清空

Session的設定、獲取、刪除類似於字典的操作

過期時間

session.permanent=True

如果設定為“ True”，則會話的生存時間為permanent_session_lifetime秒。預設值為31天。如果設定為False（預設設定），則當用戶關閉瀏覽器時，會話將被刪除。

PERMANENT_SESSION_LIFETIME 配置過期時間，單位：秒

訊號

訊號(Signal)就是兩個獨立的模組用來傳遞訊息的方式，它有一個訊息的傳送者Sender，還有一個訊息的訂閱者Subscriber。

``Flask的訊號功能是基於[Blinker](https://pypi.python.org/pypi/blinker)的，在開始此篇之前，你先要安裝blinker包pip install blinker`。

建立訊號

定義訊號需要使用到blinker這個包的 Namespace 類來建立一個名稱空間。

from blinker import Namespace
my_signals = Namespace()
login_signal = mySpace.signal('建立一個登入訊號')

訂閱訊號

使用訊號的 connect() 方法來訂閱訊號。該函式的第一個引數是訊號發出時要呼叫的函式，第二個引數是可選的，用於確定訊號的傳送端。退訂一個訊號，可以使用disconnect() 方法。

login_signal.connect(login_log)

傳送訊號

發出訊號，呼叫 send() 方法可以做到。它接受傳送端作為第一個引數，和一些推送到訊號訂閱者的可選關鍵字引數

login_signal.send()

訊號使用場景

定義一個登入的訊號，以後使用者登入進來
傳送一個登入訊號，然後能夠監聽這個訊號
在監聽到這個訊號以後，就記錄當前這個使用者登入的資訊
用訊號的方式，記錄使用者的登入資訊即登入日誌

１．編寫一個signal.py檔案建立登入訊號

from blinker import Namespace
from datetime import datetime
from flask import request, g

# 建立一個登入訊號
mySpace = Namespace()
login_signal = mySpace.signal('建立一個登入訊號')

# 監聽訊號
def login_log(sender):
    # 使用者名稱   時間 ip
    username = g.username
    now = datetime.now()
    ip = request.remote_addr
    log_data = '{username}*{now}*{ip}'.format(username=username, now=now, ip=ip)
    with open('login_log.txt', 'a')as f:
        f.write(log_data + '\n')
        
# 傳送訊號
login_signal.connect(login_log)

２．使用訊號儲存使用者登入日誌（app.py）

from flask import Flask, request, g, render_template, redirect, url_for
from my_signal import login_signal

app = Flask(__name__)

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

@app.route('/login', methods=['GET', 'POST'])
def login():
    error = None
    if request.method == 'POST':
        if request.form['username'] != 'admin' or \
                request.form['password'] != 'secret':
            error = 'Invalid credentials'
        else:
            g.username = request.form['username']
            login_signal.send() # 傳送訊號
            return redirect(url_for('index'))
    return render_template('login.html', error=error)

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

`Flask`核心訊號

官方文件

鉤子函式

在Flask的一次請求生命週期的過程中，為我們提供了多個特定的裝飾器裝飾的函式，通過這些函式可以有效的控制我們請求的生命週期，而這些函式就叫鉤子函式。

常用鉤子函式

before_first_request：Flask專案啟動後，第一次請求會執行，以後就不再執行。

before_request：請求已經到達了Flask，但是還沒有進入到具體的檢視函式之前呼叫。如果返回為非假的值，則會直接返回響應，結束請求流程，預設可以不傳，表示返回None。如果被裝飾多個函式，按照函式定義順序執行

after_request：檢視函式執行完成時呼叫，被裝飾的函式需要接受一個引數，表示response，該函式必須一個response的物件，如果修改了response，則直接返回修改後的結果。作用就是對檢視函式處理完成的響應做進一步修改然後返回。如果被裝飾多個函式，按照函式定義相反的順序執行

teardown_appcontext：在請求結束之前呼叫，被裝飾的函式需要接受一個引數，表示異常資訊

測試程式碼如下：

from flask import Flask
from flask import make_response

app = Flask(__name__)

@app.route("/index/")
def index():
    return "index"

@app.before_first_request
def before_first_request():
    print("before_first_request")

@app.before_request
def before_request_1():
    print("before_request_1")
    # return "終止請求"

@app.before_request
def before_request_2():
    print("before_request_2")
    # return "終止請求"

@app.after_request
def after_request_1(response):
    print("after_request_1")
    # return make_response("aaa")
    return response

@app.after_request
def after_request_2(response):
    print("after_request_2")
    # return make_response("aaa")
    return response

@app.teardown_appcontext
def teardown(exc):
    print("teardown")

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

"""
before_first_request
before_request_1
before_request_2
after_request_2
after_request_1
teardown
"""

其他鉤子函式

errorhandler(code_or_exception)：捕獲的響應的異常進行處理，必須要寫一個引數，來接收錯誤的資訊，如果沒有引數，就會直接報錯。最後被裝飾的函式需要返回相應的狀態碼。主要與flask.about配合使用，通過about()丟擲指定異常。

@app.errorhandler(404)
def page_not_found(error):  # 必須接受一個引數
    return 'This page does not exist', 404

context_processor：使用這個鉤子函式，必須返回一個字典。這個字典中的值在所有模版中都可以使用。

@app.context_processor
def func2():
    return {"username": "ydongy"}

流程圖：

訊息閃現

Flask 提供了一個非常簡單的方法來使用閃現系統向用戶反饋資訊。閃現系統使得在一個請求結束的時候記錄一個資訊，然後在且僅僅在下一個請求中訪問這個資料。這通常配合一個佈局模板實現。

簡單來講就是可以通過閃現訊息機制把我們後臺處理的成功或錯誤等資訊提示給使用者。

簡單閃現

在Flask中，使用flash message(閃現訊息)，具體使用的方法是flash()

flash(message, category="message")

# message:訊息提示
# category:訊息型別，比如：message，info，error,warning。根據不同的提示可以在模板中渲染不同顏色的提示資訊

在模板中接收flush()傳遞過來的資訊使用get_flashed_message()

get_flashed_messages(with_categories, category_filter)

{# with_categories: True表示一個元組，裡面是訊息型別 
  category_filter：將訊息過濾到僅匹配提供的類別。#}

一個簡單的案例：

from flask import Flask, flash, redirect, render_template, \
    request, url_for

app = Flask(__name__)
app.secret_key = 'some_secret' # 必須配置祕鑰，因為訊息閃現是基於session的

@app.route('/')
def index():
    return render_template('index.html')

@app.route('/login', methods=['GET', 'POST'])
def login():
    error = None
    if request.method == 'POST':
        if request.form['username'] != 'admin' or \
                request.form['password'] != 'secret':
            error = 'Invalid credentials'
        else:
            flash('You were successfully logged in')
            return redirect(url_for('index'))
    return render_template('login.html', error=error)

if __name__ == "__main__":
    app.run(debug=True)

獲取使用者名稱和密碼進行校驗，驗證成功呼叫flash方法跳轉到首頁，失敗停留在的登入頁並顯示錯誤資訊

這裡是 index.html 模板:

<!DOCTYPE html>
<html lang="en">
<head>
    <meta charset="UTF-8">
    <title>Index</title>
</head>
<body>
{% with messages = get_flashed_messages() %}
    {% if messages %}
        {% for message in messages %}
            <li>{{ message }}</li>
        {% endfor %}
    {% endif %}
{% endwith %}
<h3>Welcome!</h3>
<a href="{{ url_for('login') }}">login</a>
</body>
</html>

通過呼叫get_flashed_messages方法獲取到所有的訊息，然後使用for-in的迴圈顯示出每一條訊息。頁面的底部，我們放置一個超連結，用於跳轉到login頁面

這裡是login.html模板:

<!DOCTYPE html>
<html lang="en">
<head>
    <meta charset="UTF-8">
    <title>Login</title>
</head>
<body>
<form action="" method=post>
    <dl>
        <dt>Username:
        <dd><input type=text name=username value="{{ request.form.username }}">
        <dt>Password:
        <dd><input type=password name=password>
    </dl>
    <p><input type=submit value=Login>
</form>
{% if error %}
    <p><strong>Error</strong>: {{ error }}</p>
{% endif %}
</body>
</html>

啟動Flask服務，訪問http://127.0.0.1:5000

點選Login輸入使用者名稱和密碼

錯誤提交

正確提交

分類閃現

指定flash('You were successfully logged in', 'info')中的第二個引數修改閃現型別

修改模板get_flashed_messages(with_categories=true)中的with_categories=True

{% with messages = get_flashed_messages(with_categories=true) %}
    {% if messages %}
        <ul class=flashes>
            {% for category, message in messages %}
                <li class="{{ category }}">{{ message }}</li>
            {% endfor %}
        </ul>
    {% endif %}
{% endwith %}

過濾閃現訊息

修改模板get_flashed_messages(category_filter=["error"])中的category_filter引數，以過濾函式返回的結果

{% with errors = get_flashed_messages(category_filter=["error"]) %}
{% if errors %}
  <ul>
    {%- for msg in errors %}
    <li>{{ msg }}</li>
    {% endfor -%}
  </ul>
{% endif %}
{% endwith %}

資料流

當我們傳送我們的資料遠遠超過記憶體的大小並且實時地產生這些資料時，如何才能直接把他傳送給客戶端，而不需要在檔案系統中中轉呢?

答案是生成器和 Direct Response。

基本使用

下面是一個簡單的檢視函式實時生成大量的 CSV 資料，這一函式使用生成器來生成資料，並且稍後啟用這個生成器函式時，把返回值傳遞給一個 response 物件，每一個 yield 表示式直接被髮送給瀏覽器。

from flask import Response, Flask

app = Flask(__name__)

@app.route('/large.csv')
def generate_large_csv():
    def generate():
        for row in range(50000):
            line = []
            for col in range(500):
                line.append(str(col))
            if row % 1000 == 0:
                print('row: %d' % row)
            yield ','.join(line) + '\n'

    return Response(generate(), mimetype='text/csv')

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

在模板中生成流

Jinja2 模板引擎同樣支援分塊逐個渲染模板：

from flask import Response

def stream_template(template_name, **context):
    # 我們繞過了 Flask 的模板渲染函式，而是直接使用了模板物件，
    # 所以我們手動必須調update_template_context()函式來確保更新了模板的渲染上下文
    app.update_template_context(context)
    # 獲取Jinja2的模板物件
    t = app.jinja_env.get_template(template_name)
    # 獲取流式渲染模板的生成器
    rv = t.stream(context)
    # 啟用快取，這樣不會每一條都發送，而是快取滿了再發送
    rv.enable_buffering(5)
    return rv

@app.route('/my-large-page.html')
def render_large_template():
    file = open("large.csv")
    return Response(stream_template('the_template.html', csv=file.readlines()))

相關參考：
http://www.bjhee.com/flask-ad5.html
http://docs.jinkan.org/docs/flask/