Flask超強入門總結

Flask常用擴充包

• Flask-SQLalchemy：操作資料庫；
• Flask-script：插入腳本；
• Flask-migrate：管理遷移資料庫；
• Flask-Session：Session存儲方式指定；
• Flask-WTF：表單；
• Flask-Mail：郵件；
• Flask-Bable：提供國際化和本地化支援，翻譯；
• Flask-Login：認證使用者狀态；
• Flask-OpenID：認證；
• Flask-RESTful：開發REST API的工具；
• Flask-Bootstrap：內建前端Twitter Bootstrap架構；
• Flask-Moment：本地化日期和時間；
• Flask-Admin：簡單而可擴充的管理接口的架構

1.Flask基本使用

from flask import Flask

app = Flask(__name__)


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

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

參數說明

1. Flask對象初始化參數

   a.import_name：Flask程式所在的包（子產品），傳__name__ 就可以；這個參數決定Flask在通路靜态檔案時查找的路徑

   b.static_url_path：靜态檔案通路路徑，可以不傳，預設為：/ + static_folder

   c.static_folder：靜态檔案存儲的檔案夾，可以不傳，預設為static

   d.template_folder：模闆檔案存儲的檔案夾，可以不傳，預設為templates

Flask配置加載

1.從配置對象中加載

class DefaultConfig(object):
	"""
	預設配置
	"""
	SECRET_KEY = 'WDOPGHNP2UI3QHTGPEWRSQ9HY8943'

app = Flask(__name__)

app.config.from_object(DefaultConfig)

@app.route("/")
def index():
    print(app.config['SECRET_KEY'])
    return "hello world"

# 作為預設配置解除安裝程式代碼中，可以繼承
class DevelopmentConfig(DefaultConfig):
    DEBUG=True

           

2.從配置檔案中加載

建立setting.py配置檔案

app.config.from_pyfile('setting.py')

@app.route("/")
def index():
	print(app.config['SECRET_KEY'])
	return "helloworld"
           

3.從環境變量中加載

先設定環境變量

app = Flask(__name__)

'''
False 表示不安靜的處理，沒有值時報錯通知，預設為False
True 表示安靜的處理，即時沒有值也讓Flask正常的運作下去
'''
app.config.from_envvar('PROJECT_SETTING', silent=True)

@app.route("/")
def index():
    print(app.config['SECRET_KEY'])
    return "hello world"
           

Flask項目中常用的加載方式

"""
使用配置對象加載預設配置
使用環境變量加載不想出現在代碼中的敏感配置資訊
"""
def create_flask_app(config):
    """
    建立Flask應用
    :param config: 配置對象
    :return: Flask應用
    """
    app = Flask(__name__)
    app.config.from_object(config)

    # 從環境變量指向的配置檔案中讀取的配置資訊會覆寫掉從配置對象中加載的同名參數
    app.config.from_envvar("PROJECT_SETTING", silent=True)
    return app

class DefaultConfig(object):
    """預設配置"""
    SECRET_KEY = 'itcast1'

class DevelopmentConfig(DefaultConfig):
    DEBUG=True

# app = create_flask_app(DefaultConfig)
app = create_flask_app(DevelopmentConfig)

@app.route("/")
def index():
    print(app.config['SECRET_KEY'])
    return "hello world"
           

app.run參數

# 可以指定運作的主機IP位址，端口，是否開啟調試模式
app.run(host="0.0.0.0", port=5000, debug = True)
           

開發伺服器啟動方式

• 在1.0版本之後，Flask調整了開發伺服器的啟動方式，由代碼編寫app.run()語句調整為指令flask run啟動。

from flask import Flask

app = Flask(__name__)

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

# 程式中不用再寫app.run()
           

• 終端啟動

$ export FLASK_APP=helloworld
$ flask run
 * Running on http://127.0.0.1:5000/

"""
環境變量 FLASK_APP 指明flask的啟動執行個體
flask run -h 0.0.0.0 -p 8000 綁定位址 端口
flask run --help擷取幫助
生産模式與開發模式的控制
通過FLASK_ENV環境變量指明
export FLASK_ENV=production 運作在生産模式，未指明則預設為此方式
export FLASK_ENV=development 運作在開發模式
"""

$ export FLASK_APP=helloworld
$ python -m flask run
 * Running on http://127.0.0.1:5000/
           

配置Pycharm啟動

1.設定環境變量

2.舊版本Pycharm設定

2.路由和藍圖

1.查詢路由資訊

flask routes
           

在程式中擷取路由資訊

print(app.url_map)
# 如果想在程式中周遊路由資訊，可以采用如下方式
for rule in app.url_map.iter_rules():
    print('name={} path={}'.format(rule.endpoint, rule.rule))
           

執行個體

# 需求：通過通路/位址，以json的方式傳回應用内的所有路由資訊

@app.route('/')
def route_map():
    """
    主視圖，傳回所有視圖網址
    """
    rules_iterator = app.url_map.iter_rules()
    return json.dumps({rule.endpoint: rule.rule for rule in rules_iterator})
           

指定請求方式

@app.route("/itcast1", methods=["POST"])
def view_func_1():
    return "hello world 1"

@app.route("/itcast2", methods=["GET", "POST"])
def view_func_2():
    return "hello world 2"
           

2.blueprint藍圖

在Flask中，使用藍圖Blueprint來分子產品組織管理。

藍圖實際可以了解為是一個存儲一組視圖方法的容器對象，其具有如下特點：

• 一個應用可以具有多個Blueprint
• 可以将一個Blueprint注冊到任何一個未使用的URL下比如 “/user”、“/goods”
• Blueprint可以單獨具有自己的模闆、靜态檔案或者其它的通用操作方法，它并不是必須要實作應用的視圖和函數的
• 在一個應用初始化時，就應該要注冊需要使用的Blueprint
• 但是一個Blueprint并不是一個完整的應用，它不能獨立于應用運作，而必須要注冊到某一個應用中。

使用方式

# 建立一個藍圖對象
user_bp=Blueprint('user',__name__)

# 在這個藍圖對象上進行操作,注冊路由,指定靜态檔案夾,注冊模版過濾器
@user_bp.route('/')
def user_profile():
    return 'user_profile'

# 在應用對象上注冊這個藍圖對象
app.register_blueprint(user_bp)

# 指定藍圖的url字首
app.register_blueprint(user_bp, url_prefix='/user')
app.register_blueprint(goods_bp, url_prefix='/goods')

# 藍圖内部靜态檔案
# 現在就可以使用/admin/static_admin/<filename>通路static_admin目錄下的靜态檔案了。
admin = Blueprint("admin",__name__,static_folder='static_admin')
app.register_blueprint(admin,url_prefix='/admin')
# 也可通過static_url_path改變通路路徑
admin = Blueprint("admin",__name__,static_folder='static_admin',static_url_path='/lib')
app.register_blueprint(admin,url_prefix='/admin')

# 藍圖内部模闆目錄
admin = Blueprint('admin',__name__,template_folder='my_templates')
           

3.請求與響應

request相關内容

如果想要擷取其他地方傳遞的參數，可以通過Flask提供的request對象來讀取。

屬性	說明	類型
data	記錄請求的資料，并轉換為字元串	*
form	記錄請求中的表單資料	MultiDict
args	記錄請求中的查詢參數	MultiDict
cookies	記錄請求中的cookie資訊	Dict
headers	記錄請求中的封包頭	EnvironHeaders
method	記錄請求使用的HTTP方法	GET/POST
url	記錄請求的URL位址	string
files	記錄請求上傳的檔案	*

例如 想要擷取請求/articles?channel_id=1中channel_id的參數，可以按如下方式使用：

from flask import request

@app.route('/articles')
def get_articles():
    channel_id = request.args.get('channel_id')
    return 'you wanna get articles of channel {}'.format(channel_id)

from flask import request


# 上傳圖檔
@app.route('/upload', methods=['POST'])
def upload_file():
    f = request.files['pic']
    # with open('./demo.png', 'wb') as new_file:
    #     new_file.write(f.read())
    f.save('./demo.png')
    return 'ok'
           

處理請求

"""
 URL路徑參數（動态路由）
"""
@app.route('/users/<user_id>')
def user_info(user_id):
    print(type(user_id))
    return 'hello user {}'.format(user_id)


"""
此處的<>即是一個轉換器，預設為字元串類型
即将該位置的資料以字元串格式進行比對、并以字元串為資料類型類型、 user_id為參數名傳入視圖。

Flask也提供其他類型的轉換器

DEFAULT_CONVERTERS = {
    'default':          UnicodeConverter,
    'string':           UnicodeConverter,
    'any':              AnyConverter,
    'path':             PathConverter,
    'int':              IntegerConverter,
    'float':            FloatConverter,
    'uuid':             UUIDConverter,
}
"""
@app.route('/users/<int:user_id>')
def user_info(user_id):
    print(type(user_id))
    return 'hello user {}'.format(user_id)

@app.route('/users/<int(min=1):user_id>')
def user_info(user_id):
    print(type(user_id))
    return 'hello user {}'.format(user_id)
           

自定義轉換器

"""
應用場景：
	如果遇到需要比對提取/sms_codes/18512345678中的手機号資料，Flask内置的轉換器就無法滿足需求，此時需要自定義轉換器。
"""
# 1.建立轉換器類，儲存比對時的正規表達式
from werkzeug.routing import BaseConverter

class MobileConverter(BaseConverter):
    """
    手機号格式
    """
    # 注意regex名字固定
    regex = r'1[3-9]\d{9}'

# 2.将自定義的轉換器告知Flask應用
app = Flask(__name__)
# 将自定義轉換器添加到轉換器字典中，并指定轉換器使用時名字為: mobile
app.url_map.converters['mobile'] = MobileConverter

# 3.在使用轉換器的地方定義使用
@app.route('/sms_codes/<mobile:mob_num>')
def send_sms_code(mob_num):
    return 'send sms code to {}'.format(mob_num)
           

response相關内容

# 傳回模闆
from flask import render_template

@app.route('/')
def index():
    mstr = 'Hello 黑馬程式員'
    mint = 10
    return render_template('index.html', my_str=mstr, my_int=mint)


# 重定向
from flask import redirect

@app.route('/demo2')
def demo2():
    return redirect('http://www.itheima.com')


# 傳回json
from flask import jsonify

@app.route('/demo3')
def demo3():
    json_dict = {
        "user_id": 10,
        "user_name": "laowang"
    }
    return jsonify(json_dict)


"""
自定義狀态碼和響應頭
"""
# 1.元組的方式
# 可以傳回一個元組，這樣的元組必須是 (response, status, headers) 的形式，且至少包含一個元素。 
# status 值會覆寫狀态代碼， headers 可以是一個清單或字典，作為額外的消息标頭值。

@app.route('/demo4')
def demo4():
    # return '狀态碼為 666', 666
    # return '狀态碼為 666', 666, [('Itcast', 'Python')]
    return '狀态碼為 666', 666, {'Itcast': 'Python'}


# make_response方式
@app.route('/demo5')
def demo5():
    resp = make_response('make response測試')
        resp.headers[“Itcast”] = “Python”
        resp.status = “404 not found”
    return resp
           

cookie和session

"""
cookie的相關設定
"""
# 設定cookie,并設定有效期
from flask import Flask, make_response

app = Flask(__name__)

@app.route('/cookie')
def set_cookie():
    resp = make_response('set cookie ok')
    resp.set_cookie('username', 'itcast'， max_age=3600)
    return resp


# 讀取cookie
from flask import request

@app.route('/get_cookie')
def get_cookie():
    resp = request.cookies.get('username')
    return resp

# 删除cookie
from flask import request

@app.route('/delete_cookie')
def delete_cookie():
    response = make_response('hello world')
    response.delete_cookie('username')
    return response
           

"""
session相關設定
必須依賴SECRET_KEY
"""

from flask import session

# 設定
@app.route('/set_session')
def set_session():
    session['username'] = 'itcast'
    return 'set session ok'

# 擷取
@app.route('/get_session')
def get_session():
    username = session.get('username')
    return 'get session username {}'.format(username)
           

4.異常處理

• abort 方法

  抛出一個給定狀态代碼的 HTTPException 或者 指定響應，例如想要用一個頁面未找到異常來終止請求，你可以調用 abort(404)。

• 參數：

  code – HTTP的錯誤狀态碼

• 捕獲錯誤

"""
errorhandler 裝飾器
	注冊一個錯誤處理程式，當程式抛出指定錯誤狀态碼的時候，就會調用該裝飾器所裝飾的方法
參數：
	code_or_exception – HTTP的錯誤狀态碼或指定異常
"""
# 如統一處理狀态碼為500的錯誤給使用者友好的提示：
@app.errorhandler(500)
def internal_server_error(e):
    return '伺服器搬家了'

# 捕獲指定異常
@app.errorhandler(ZeroDivisionError)
def zero_division_error(e):
    return '除數不能為0'
           

5.請求鈎子與上下文

"""
請求鈎子
"""
# 在第一次請求之前調用，可以在此方法内部做一些初始化操作
@app.before_first_request
def before_first_request():
    print("before_first_request")


# 在每一次請求之前調用，這時候已經有請求了，可能在這個方法裡面做請求的校驗
# 如果請求的校驗不成功，可以直接在此方法中進行響應，直接return之後那麼就不會執行視圖函數
@app.before_request
def before_request():
    print("before_request")
    # if 請求不符合條件:
    #     return "laowang"


# 在執行完視圖函數之後會調用，并且會把視圖函數所生成的響應傳入,可以在此方法中對響應做最後一步統一的處理
@app.after_request
def after_request(response):
    print("after_request")
    response.headers["Content-Type"] = "application/json"
    return response


# 請每一次請求之後都會調用，會接受一個參數，參數是伺服器出現的錯誤資訊
@app.teardown_request
def teardown_request(response):
    print("teardown_request")


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

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

           

請求上下文（request context）

在 flask 中，可以直接在視圖函數中使用 request 這個對象進行擷取相關資料，而 request 就是請求上下文的對象，儲存了目前本次請求的相關資料

請求上下文對象有：request、session

• request
  • 封裝了HTTP請求的内容，針對的是http請求。舉例：user = request.args.get(‘user’)，擷取的是get請求的參數。
• session
  • 用來記錄請求會話中的資訊，針對的是使用者資訊。舉例：session[‘name’] = user.id，可以記錄使用者資訊。還可以通過session.get(‘name’)擷取使用者資訊。

應用上下文

它的字面意思是 應用上下文，但它不是一直存在的，它隻是request context 中的一個對 app 的代理(人)，所謂local proxy。

它的作用主要是幫助 request 擷取目前的應用，它是伴 request 而生，随 request 而滅的。

應用上下文對象有：current_app，g

• current_app
  • 應用程式上下文,用于存儲應用程式中的變量，可以通過current_app.name列印目前app的名稱，也可以在current_app中存儲一些變量(應用的啟動腳本是哪個檔案，啟動時指定了哪些參數,加載了哪些配置檔案，導入了哪些配置,連了哪個資料庫,應用跑再哪個機器上，IP多少，記憶體多大)

"""
應用上下文
current_app 就是目前運作的flask app，在代碼不友善直接操作flask的app對象時
可以操作current_app就等價于操作flask app對象
"""
from flask import Flask, current_app

app1 = Flask(__name__)
app2 = Flask(__name__)

# 以redis用戶端對象為例
# 用字元串表示建立的redis用戶端
# 為了友善在各個視圖中使用，将建立的redis用戶端對象儲存到flask app中，
# 後續可以在視圖中使用current_app.redis_cli擷取
app1.redis_cli = 'app1 redis client'
app2.redis_cli = 'app2 redis client'

@app1.route('/route11')
def route11():
    return current_app.redis_cli

@app1.route('/route12')
def route12():
    return current_app.redis_cli

@app2.route('/route21')
def route21():
    return current_app.redis_cli

@app2.route('/route22')
def route22():
    return current_app.redis_cli
           

• g對象
  • g 作為 flask 程式全局的一個臨時變量，充當中間媒介的作用，我們可以通過它在一次請求調用的多個函數間傳遞一些資料。每次請求都會重設這個變量。

from flask import Flask, g

app = Flask(__name__)

def db_query():
    user_id = g.user_id
    user_name = g.user_name
    print('user_id={} user_name={}'.format(user_id, user_name))

@app.route('/')
def get_user_profile():
    g.user_id = 123
    g.user_name = 'itcast'
    db_query()
    return 'hello world'


# 執行個體
# 需求：建構認證機制，對于特定視圖可以提供強制要求使用者登入的限制，對于所有視圖，無論是否強制要求使用者登入，都可以在視圖中嘗試擷取使用者認證後的身份資訊
from flask import Flask, abort, g

app = Flask(__name__)

@app.before_request
def authentication():
    """
    利用before_request請求鈎子，在進入所有視圖前先嘗試判斷使用者身份
    :return:
    """
    # TODO 此處利用鑒權機制（如cookie、session、jwt等）鑒别使用者身份資訊
    # if 已登入使用者，使用者有身份資訊
    g.user_id = 123
    # else 未登入使用者，使用者無身份資訊
    # g.user_id = None

def login_required(func):
    def wrapper(*args, **kwargs):
        if g.user_id is not None:
            return func(*args, **kwargs)
        else:
            abort(401)

    return wrapper

@app.route('/')
def index():
    return 'home page user_id={}'.format(g.user_id)

@app.route('/profile')
@login_required
def get_user_profile():
    return 'user profile page user_id={}'.format(g.user_id)
           

app_context 與 request_context

• app_context
  • 為我們提供了應用上下文環境，允許我們在外部使用應用上下文current_app、g

可以通過with語句進行使用

>>> from flask import Flask
>>> app = Flask('')
>>> app.redis_cli = 'redis client'
>>> 
>>> from flask import current_app
>>> current_app.redis_cli   # 錯誤，沒有上下文環境
報錯
>>> with app.app_context():  # 借助with語句使用app_context建立應用上下文
...     print(current_app.redis_cli)
...
redis client
           

request_context

• request_context為我們提供了請求上下文環境，允許我們在外部使用請求上下文request、session

可以通過with語句進行使用

>>> from flask import Flask
>>> app = Flask('')
>>> request.args  # 錯誤，沒有上下文環境
報錯
>>> environ = {'wsgi.version':(1,0), 'wsgi.input': '', 'REQUEST_METHOD': 'GET', 'PATH_INFO': '/', 'SERVER_NAME': 'itcast server', 'wsgi.url_scheme': 'http', 'SERVER_PORT': '80'}  # 模拟解析用戶端請求之後的wsgi字典資料
>>> with app.request_context(environ):  # 借助with語句使用request_context建立請求上下文
...     print(request.path)
...   
/
           

6.Flask-Restful

# 安裝
pip install flask-restful

# Hello World
from flask import Flask
from flask_restful import Resource, Api

app = Flask(__name__)
api = Api(app)

class HelloWorldResource(Resource):
    def get(self):
        return {'hello': 'world'}

        def post(self):
        return {'msg': 'post hello world'}

api.add_resource(HelloWorldResource, '/')

# 此處啟動對于1.0之後的Flask可有可無
if __name__ == '__main__':
    app.run(debug=True)
           

關于視圖

"""
用endpoint參數為路由起名
"""
api.add_resource(HelloWorldResource, '/', endpoint='HelloWorld')


"""
藍圖中使用
"""
from flask import Flask, Blueprint
from flask_restful import Api, Resource

app = Flask(__name__)

user_bp = Blueprint('user', __name__)

user_api = Api(user_bp)

class UserProfileResource(Resource):
    def get(self):
        return {'msg': 'get user profile'}

user_api.add_resource(UserProfileResource, '/users/profile')

app.register_blueprint(user_bp)


"""
裝飾器
"""
# 為類視圖中的所有方法添加裝飾器
def decorator1(func):
    def wrapper(*args, **kwargs):
        print('decorator1')
        return func(*args, **kwargs)
    return wrapper


def decorator2(func):
    def wrapper(*args, **kwargs):
        print('decorator2')
        return func(*args, **kwargs)
    return wrapper


class DemoResource(Resource):
    method_decorators = [decorator1, decorator2]

    def get(self):
        return {'msg': 'get view'}

    def post(self):
        return {'msg': 'post view'}

# 為類視圖中不同的方法添加不同的裝飾器
class DemoResource(Resource):
    method_decorators = {
        'get': [decorator1, decorator2],
        'post': [decorator1]
    }

    # 使用了decorator1 decorator2兩個裝飾器
    def get(self):
        return {'msg': 'get view'}

    # 使用了decorator1 裝飾器
    def post(self):
        return {'msg': 'post view'}

    # 未使用裝飾器
    def put(self):
        return {'msg': 'put view'}
           

關于請求處理

Flask-RESTful 提供了RequestParser類，用來幫助我們檢驗和轉換請求資料。

"""
使用步驟：
1.建立RequestParser對象
2.向RequestParser對象中添加需要檢驗或轉換的參數聲明
3.使用parse_args()方法啟動檢驗處理
4.檢驗之後從檢驗結果中擷取參數時可按照字典操作或對象屬性操作
"""
from flask_restful import reqparse

parser = reqparse.RequestParser()
parser.add_argument('rate', type=int, help='Rate cannot be converted', location='args')
parser.add_argument('name')
args = parser.parse_args()
args.rate
# 或者
args['rate']


"""
參數說明
required: 描述請求是否一定要攜帶對應參數，預設值為False

help: 參數檢驗錯誤時傳回的錯誤描述資訊

action: 描述對于請求參數中出現多個同名參數時的處理方式
	- action='store' 保留出現的第一個， 預設
	- action='append' 以清單追加儲存所有同名參數的值

type: 描述參數應該比對的類型，可以使用python的标準資料類型string、int;也可使用Flask-RESTful提供的檢驗方法，還可以自己定義

location: 描述參數應該在請求資料中出現的位置
"""
# required
class DemoResource(Resource):
    def get(self):
        rp = RequestParser()
        rp.add_argument('a', required=False)
        args = rp.parse_args()
        return {'msg': 'data={}'.format(args.a)}

# help
rp.add_argument('a', required=True, help='missing a param')

# action
rp.add_argument('a', required=True, help='missing a param', action='append')

# type
rp.add_argument('a', type=int, required=True, help='missing a param', action='append')
# type   
"""
flask-restful提供的檢驗方法
natural: 自然數0、1、2、3...
positive: 正整數 1、2、3...
boolean:
url:
"""
# regex(指定正規表達式)
from flask_restful import inputs
rp.add_argument('a', type=inputs.regex(r'^\d{2}&'))
# int_range(low ,high) 整數範圍
rp.add_argument('a', type=inputs.int_range(1, 10))
"""
自定義校驗
"""
def mobile(mobile_str):
    """
    檢驗手機号格式
    :param mobile_str: str 被檢驗字元串
    :return: mobile_str
    """
    if re.match(r'^1[3-9]\d{9}$', mobile_str):
        return mobile_str
    else:
        raise ValueError('{} is not a valid mobile'.format(mobile_str))

rp.add_argument('a', type=mobile)

# location
# Look only in the POST body
parser.add_argument('name', type=int, location='form')

# Look only in the querystring
parser.add_argument('PageSize', type=int, location='args')

# From the request headers
parser.add_argument('User-Agent', location='headers')

# From http cookies
parser.add_argument('session_id', location='cookies')

# From json
parser.add_argument('user_id', location='json')

# From file uploads
parser.add_argument('picture', location='files')

# 也可以指定多個位置
parser.add_argument('text', location=['headers', 'json'])
           

關于響應處理

Flask-RESTful 提供了marshal工具，用來幫助我們将資料序列化為特定格式的字典資料，以便作為視圖的傳回值。

from flask_restful import Resource, fields, marshal_with

resource_fields = {
    'name': fields.String,
    'address': fields.String,
    'user_id': fields.Integer
}

class Todo(Resource):
    @marshal_with(resource_fields, envelope='resource')
    def get(self, **kwargs):
        return db_get_todo()

# 示例
# 用來模拟要傳回的資料對象的類
class User(object):
    def __init__(self, user_id, name, age):
        self.user_id = user_id
        self.name = name
        self.age = age

resoure_fields = {
        'user_id': fields.Integer,
        'name': fields.String
    }

class Demo1Resource(Resource):
    @marshal_with(resoure_fields, envelope='data1')
    def get(self):
        user = User(1, 'itcast', 12)
        return user

class Demo2Resource(Resource):
    def get(self):
        user = User(1, 'itcast', 12)
        return marshal(user, resoure_fields, envelope='data2')