Flask詳解

配置文件

Flask 中所有的配置文件可以通過Flask(_name_).config查看。實際上是一個flask.config.Config對象

from flask import Flask
app = Flask(__name__)
print(app.config)

默認配置文件

{
    'DEBUG':                                get_debug_flag(default=False)
    '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,
}

修改配置文件的方式

基於類

class DevConfig:
    DEBUG = True
	# ......
    
   
app.config.from_object(DevConfig)

其他修改方式

# 直接修改
app.config['DEBUG'] = True


# py文件
app.config.from_pyfile("settings.py")
# settings.py文件
DEBUG = True


# 環境變量的值為python文件名稱名稱，內部調用from_pyfile方法
pp.config.from_envvar("環境變量名稱")


# json文件
app.config.from_json("setting.json")
# setting.json 文件
{"DEBUG": true}


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

路由系統

url反向生成

endpoint 反向生成url，endpoint默認就是函數名

from flask import url_for

@app.route('/index', methods=['GET', "POST"], endpoint='xxx')
def index():
    print(url_for('xxx'))
    ....
    

url動態傳參

@app.route('/index/<int:nid>')			# 整數
def index(nid):
    print(nid,type(nid))
    return 'index'

其他常見寫法

@app.route('/index/<int:nid>')			# 整數
@app.route('/index/<float:nid>')		# 小數
@app.route('/index/<path:nid>')			# 路徑
@app.route('/index/<uuid:nid>')			# uuid
@app.route('/index/<nid>')			# 字符串/@app.route('/index/<str:nid>')
@app.route('/index')				# 啥都不傳

如果想反向生成url就需要額外傳參

print(url_for('index', nid=123))		# 傳入的數據必須和url匹配的類型一樣

所有路由的對應關系

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

不采用裝飾器的寫法

def index():
    return 'xxx'

app.add_url_rule(rule='/index', view_func=index)

參數說明

rule					url路徑
view_func				視圖函數
defaults=None			        url傳參{'k':'v'}
endpoint=None			        別名,為None時則等於函數名
methods					允許的請求方式
strict_slashes=True		        對url最后的 / 符號是否嚴格要求
redirect_to=None		        重定向地址
subdomain=None			        子域名訪問

自定義路由匹配

from werkzeug.routing import BaseConverter
# 自定義一個類，繼承BaseConverter
class MyRouter(BaseConverter):
    regex = '\d+'
    def to_python(self, value):
        return int(value)

    def to_url(self, value):
        return str(value)

# 添加到轉換器中
app.url_map.converters['m_r'] = MyRouter

# 使用自定義路由匹配
@app.route('/index/<m_r:nid>')
def index(nid):
    print(nid)
    return 'xxx'

萬能版

from werkzeug.routing import BaseConverter
class MyRouter(BaseConverter):
    def __init__(self, map, regex):
        super().__init__(map)
        self.regex = regex

    def to_python(self, value):
        return int(value)

    def to_url(self, value):
        return str(value)


app.url_map.converters['MyRouter'] = MyRouter


@app.route('/index/<MyRouter("\d{3}"):nid>')
def index(nid):
    print(nid)
    return 'xxx'

視圖

FBV

from flask import Flask

app = Flask(__name__)


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


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

加裝飾器

from functools import wraps
def decorate(f):
    @wraps(f)
    def wrapper(*args, **kwargs):
        print('前')
        rest = f(*args, **kwargs)
        print('后')
        return rest

    return wrapper

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

CBV

from flask import Flask
from flask import views
from functools import wraps

app = Flask(__name__)


def decorate(f):
    @wraps(f)
    def wrapper(*args, **kwargs):
        print('前')
        rest = f(*args, **kwargs)
        print('后')
        return rest

    return wrapper


class IndexView(views.MethodView):
    methods = ['GET', 'POST']
    decorators = [decorate, ]

    def get(self, *args, **kwargs):
        return 'GET'

    def post(self, *args, **kwargs):
        return 'POST'


app.add_url_rule('/index', view_func=IndexView.as_view('index'))

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

請求&響應

from flask import request
from flask import make_response
from flask import jsonify
from flask import redirect
from flask import render_template
from werkzeug import secure_filename

@app.route('/index/<int:nid>', methods=['GET', "POST"])
def index(nid):
    # 請求相關信息
    # request.method
    # request.args
    # request.form
    # request.values
    # request.data
    # request.json
    # request.cookies
    # 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')
    # return jsonify({'k':'v'})
    
    # 定制cookie，響應頭
    # response 是 flask.wrappers.Response 類型
    # response = make_response(render_template('index.html'))
    # response.delete_cookie('key')
    # response.set_cookie('key', 'value')
    # response.headers['X-Something'] = 'A value'
    # return response

模板渲染(jinja2)

jinja2 模板語法和 DjangoTemplates 相似，但是和Django模板相比更貼近Python的語法

官方文檔 中文文檔

定義宏

jinja2 中的宏類似python中的函數

# 定義一個宏
{% macro input(name, type='text', value='') -%}
<input type="{{ type }}" name="{{ name }}" value="{{ value }}">
{%- endmacro %}

# 調用宏
{{ input('username') }}
{{ input('password', type='password') }}

自定義標簽和過濾器(特殊裝飾器)

@app.template_global()
def func(v1, v2):
    return v1 + v2


@app.template_filter()
def func2(v1, v2, v3):
    return v1 + v2 + v3


@app.route('/index')
def index():
    return render_template('index.html', a=1)

{{ func(1, 2) }}
{{ a | func2(2, 3) }}

session

在 Flask 中操作 session 的方式和操作 Python 字典的方式幾乎一樣，跟Django的session一樣，Flask也有SECRET_KEY，但是Flask中默認SECRET_KEY=None需要手動指定

app.secret_key = 'xxx'

或者直接在配置文件中指定

class DevConfig:
    DEBUG = True
    SECRET_KEY = 'xxx'
	# ......
app.config.from_object(DevConfig)

• 當請到來時，Flask會讀取cookie中session這個key對應的value，並且將value解密、反序列化為Python的字典，然后放到內存中，供視圖函數使用
• 當請求結束時，Flask會讀取內存中session字典的值，然后再進行序列化、加密，最后寫入到cookie中

session常用操作

from flask import session

session.update({"is_login":True, 'user':'xxx'})
session.get('is_login')
del session['xxx']
session.items()
session.clear()

閃現(flash)

flash一個基於session實現的用於保存數據的集合，讀取時使用pop將其移出，所以閃現的特點是：使用一次就刪除。

from flask import flash, get_flashed_messages

@app.route('/page1')
def page1():
    flash(1,'info')
    flash(11,'error')
    flash([1,2,3],'xxx')
    return '1'


@app.route('/page2')
def page2():
    print(get_flashed_messages(category_filter=['info']))
    return '2'

閃現機制內部源碼

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
    )
    

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

中間件

一個簡單的flask示例

from flask import Flask

app = Flask(__name__)


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


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

當程序執行時，就會執行 app.run() 方法，在run方法內部又執行了run_simple，所以run_simple就是整個程序的入口，源碼如下

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

    # debug passed to method overrides all other sources
    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
    # pick the first value that's not None (0 is allowed)
    port = int(next((p for p in (port, sn_port) if p is not None), _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:
        # reset the first request information if the development server
        # reset normally.  This makes it possible to restart the server
        # without reloader and that stuff from an interactive shell.
        self._got_first_request = False

當請求進來時，就會執行app()，此時就會調用app.__call__，__call__源碼如下

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)

通過源碼分析，我們可以人為的添加一個中間件，但是基本上我們不用這種方式的中間件。但是在開發中一般用before_request和after_request裝飾器，在執行業務代碼之前或之后進行一些操作

from flask import Flask

app = Flask(__name__)


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


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

    def __call__(self, *args, **kwargs):
        print('請求來了')
        rest = self.wsgi_app(*args, **kwargs)
        print('請求走了')
        return rest


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

藍圖(BluePrint)

作用：給開發者提供目錄結構

小藍圖代碼示例

• 子域名控制(任選其一)
  • user = Blueprint('user', _name_, subdomain='user')
  • app.register_blueprint(user, subdomain='user')
• url前綴(任選其一)
  • app.register_blueprint(ac, url_prefix='/account')
  • ac= Blueprint('ac', _name_, url_prefix='/account')

大型藍圖類似Django代碼示例

特殊裝飾器

before_request，after_request

類似Django process_request process_response。用戶登陸驗證例子

# coding=utf-8
from flask import Flask
from flask import request
from flask import session
from flask import render_template
from flask import redirect
from flask import url_for

app = Flask(__name__)
app.secret_key = 'dsadsadasdsadsa'

@app.before_request
def before_request():
    if request.path != url_for('login'):
        if not session.get('is_login'):
            return redirect('login')


@app.after_request
def after_request(response):
    print(f'用戶訪問了：{request.path}')
    return response


@app.route('/login', methods=['GET', 'POST'])
def login():
    if request.method == 'GET':
        return render_template('login.html')

    user = request.form.get('username')
    pwd = request.form.get('password')
    if user == 'test' and pwd == '123':
        session.update({"is_login": True, 'user': user})
        return redirect('index')
    return redirect('login')


@app.route('/index')
def index():
    user = session.get('user')
    return f"首頁! 歡迎 {user}"


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

如果有多個before_request，after_request，before_request按定義的順序執行，after_request倒敘執行

before_first_request

只有第一個請求過來時才執行，剩下的請求不執行

@app.before_first_request
def first():
    print('xxx')

errorhandler

@app.errorhandler(404)
def page404(msg):
    print(msg)
    return '404'