Flask高级
關于Flask啟動，和請求處理
#關于后端服務的啟動，無非就是啟動實現了WSGI協議的socket
#關于flask啟動的無非就是下面兩段代碼
#加載配置，創建Flask對象
app = Flask(__name__)
#這個是啟動werkzeug服務
app.run()
下面我們來分析上面兩段代碼
app.run()
run_simple(host, port, self, **options)找到起服務的核心代碼
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 "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()
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
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:
#############################################
#開啟werkzeug服務
run_simple(host, port, self, **options)
finally:
self._got_first_request = False
請求-app.__call__()
上面我們已經啟動了wsgi服務
run_simple(host, port, self, **options)
當來一個請求來的時候wsgi就會執行一次self()
當執行self()就相當于執行了app.__call__()
app.__call__()
def __call__(self, environ, start_response):
#來一次請求就執行一次這個函數
return self.wsgi_app(environ, start_response)
wsgi_app(environ, start_response)
#首先講下函數的三個參數，self就是app,environ就行請求相關的數據，start_response是一個回調函數，是wsgi的用于返回一個響應行
def wsgi_app(self, environ, start_response):
#首先總覽一下最核心的代碼
#實例化一個ctx對象，里面封裝了request和session等信息
ctx = self.request_context(environ)
error = None
try:
try:
#將ctx存到LOCAL對象的storage，存的格式為： storage={"執行id":{'stack':[ctx,]}}
#這個也就是flask的request的奇怪用法
ctx.push()
#這里主要實現的是請求擴展
#如before_first_request,信號，和請求拓展這些函數規則的定制
response = self.full_dispatch_request()
except Exception as e:
error = e
response = self.handle_exception(e)
except: # noqa: B001
error = sys.exc_info()[1]
raise
#返回響應信息
return response(environ, start_response)
finally:
if self.should_ignore_error(error):
error = None
#刪除ctx
ctx.auto_pop(error)
其他
藍圖-flask項目架構
藍圖的主要功能就是視圖分離，不然所有視圖寫在一個文件會顯得很雜亂
使用三步
創建藍圖
from flask import Blueprint
#view
user_bp = Blueprint('user',__name__)
利用藍圖創建路由關系
#view
@bp.route('/login/')
def login():
return "login"
注冊藍圖
#app
app.register_blueprint(bp)
注意
使用url_for,前面要加上藍圖名
url_for('zx.login')
實例
項目目錄：
-templates
-static
-views
-user.py
-order.py
-app.py
views/user.py
from flask import Blueprint
# 1 創建藍圖
user_bp = Blueprint('user',__name__)
# 2 利用藍圖創建路由關系
@user_bp.route('/login/')
def login():
return "login"
@user_bp.route('/logout/')
def logout():
return "logout"
views/order.py
from flask import Blueprint
order_bp = Blueprint('order',__name__)
@order_bp.route('/add_order/')
def add_order():
return "add_order"
@order_bp.route('/modify_order/')
def modify_order():
return "modify_order"
app.py
from flask import Flask
from views.user import user_bp
from views.order import order_bp
app = Flask(__name__)
# 3 注冊藍圖
app.register_blueprint(user_bp)
app.register_blueprint(order_bp)
if __name__ == '__main__':
app.run()
g對象
g對象使用周期是一次請求，先設置值，在取值
from flask import g
#設置值
g.name = "zx"
#取值
g.name
信號
安裝
pip install blinker
內置信號
request_started = _signals.signal('request-started') # 請求到來前執行
request_finished = _signals.signal('request-finished') # 請求結束后執行
before_render_template = _signals.signal('before-render-template') # 模板渲染前執行
template_rendered = _signals.signal('template-rendered') # 模板渲染后執行
got_request_exception = _signals.signal('got-request-exception') # 請求執行出現異常時執行
request_tearing_down = _signals.signal('request-tearing-down') # 請求執行完畢后自動執行（無論成功與否）
appcontext_tearing_down = _signals.signal('appcontext-tearing-down')# 應用上下文執行完畢后自動執行（無論成功與否）
appcontext_pushed = _signals.signal('appcontext-pushed') # 應用上下文push時執行
appcontext_popped = _signals.signal('appcontext-popped') # 應用上下文pop時執行
message_flashed = _signals.signal('message-flashed') # 調用flask在其中添加數據時，自動觸發
使用信號
# 往信號中注冊函數
def func(*args,**kwargs):
print('觸發型號',args,kwargs)
#這個是內置信號
signals.request_started.connect(func)
一個源碼中的內置信號
def full_dispatch_request(self):
"""Dispatches the request and on top of that performs request
pre and postprocessing as well as HTTP exception catching and
error handling.
.. versionadded:: 0.7
"""
self.try_trigger_before_first_request_functions()
try:
#這里看到沒有
request_started.send(self)
rv = self.preprocess_request()
if rv is None:
rv = self.dispatch_request()
except Exception as e:
rv = self.handle_user_exception(e)
return self.finalize_request(rv)
自定義信號
from flask import Flask, current_app, flash, render_template
from flask.signals import _signals
app = Flask(import_name=__name__)
# 自定義信號
xxxxx = _signals.signal('xxxxx')
def func(sender, *args, **kwargs):
print(sender)
# 自定義信號中注冊函數
xxxxx.connect(func)
@app.route("/x")
def index():
# 觸發信號
xxxxx.send('123123', k1='v1')
return 'Index'
if __name__ == '__main__':
app.run()
自定義session
我們知道Django的session是存在數據庫的，而flask的默認是通過cookie實現的，那我們如果想要存在數據庫怎么辦呢？
安裝
pip install flask-session
使用
from flask import Flask,session
from flask_session import RedisSessionInterface
import redis
app = Flask(__name__)
conn=redis.Redis(host='127.0.0.1',port=6379)
#use_signer是否對key簽名,默認為false
#key_prefix為存儲的k值的前綴
#permanent默認為True，瀏覽器關閉cookie失效
app.session_interface=RedisSessionInterface(conn,key_prefix='zx')
@app.route('/')
def hello_world():
session['name']='zx125'
print(session['name'])
return 'Hello World!'
if __name__ == '__main__':
app.run()
自定義命令
安裝
pip install flask-script
使用
from flask_script import Manager
app = Flask(__name__)
manager=Manager(app)
...
if __name__ == '__main__':
manager.run()
#以后在執行，直接：python3 manage.py runserver
#python3 manage.py runserver --help
參考鏈接
https://www.cnblogs.com/xiaoyuanqujing/protected/articles/11715484.html
總結
- 上一篇： applicationContext-r
- 下一篇： eclipse卸载自带maven