面試的時候聊： 1. Flask中令你影響深刻的是什么？ - 路由系統 - 裝飾器，帶參數的裝飾器 - 額外裝飾器 - 特殊的裝飾器 2. 有沒有遇到印象深刻： - 本地線程 - 最大共享數（文檔中寫的是最大共享數，但是看源碼實現時發現pymysql.threadsafety=1有關），無用。
1. flask知識點： - flask依賴wsgi，實現wsgi協議的模塊：wsgiref(django),werkzeug,uwsgi
- 創建Flask對象 app = Flask(__name__)
def __init__(self, import_name, static_url_path=None,
static_folder='static', template_folder='templates',
instance_path=None, instance_relative_config=False,
root_path=None):
- 當前模塊名 - 靜態文件文件前綴 - 靜態文件文件夾位置 - 模板路徑 - 配置文件尋找位置： from flask import Flask app = Flask(__name__,instance_path=None, instance_relative_config=True) #只有當 引用文件的方式是 配置文件時，instance這兩個參數才有用 #instance_path是指定 從哪個位置開始找settings文件
#__name__ 是當前模塊名，當模塊被直接運行時模塊名為 __main__ 。這句話的意思就是，當模塊被直接運行時，以下代碼塊將被運行，當模塊是被導入時，代碼塊不被運行。
app.config.from_pyfile('settings.py') # C:\Users\Administrator\PycharmProjects\s6day120\1.實例化補充 if __name__ == '__main__': app.run() - 配置文件 - 推薦使用對象方式 方式一： app.config['SESSION_COOKIE_NAME'] = 'session_lvning' # 方式二： app.config.from_pyfile('settings.py') settings.py ： XXX=123 方式三： import os os.environ['FLAKS-SETTINGS'] = 'settings.py' app.config.from_envvar('FLAKS-SETTINGS') 方式四： app.config.from_object('settings.DevConfig') settings.py ： class BaseConfig(object): NNN = 123 class TestConfig(BaseConfig): DB = '127.0.0.1' class DevConfig(BaseConfig): DB = '192.168.1.1' class ProConfig(BaseConfig): DB = '47.18.1.1' - 路由系統 - 通過帶參數的裝飾器實現的路由關系 注意：其他的裝飾器要寫在路由裝飾器下面 - 兩種形式： 第一種： @app.route('/xxxx') def index(): return "Index" 第二種： def index(): return "Index" app.add_url_rule('/xxx', "n1", index) #n1 是別名 - 將函數和url封裝到一個 Rule對象 - 將Role對象添加到 app.url_map（Map對象） - 參數： (url路徑，endpoint，視圖函數名，method=[]，default，redirect_to，strict_slashes，subdomain) endpoint 是路由的別名，不寫時默認是被裝飾的函數名； 反向生成用url_for 來實現 url_for("aaa",**dic)或者 url_for("aaa",x=1) method 允許的請求方式 /index/<int:nid> 不寫類型的時候默認是 字符串 def index(nid): print(nid) defaults=None, 默認值,當URL中無參數，函數需要參數時，使用defaults={'k':'v'}為函數提供參數 strict_slashes=None, 對URL最后的 / 符號是否嚴格要求， redirect_to 直接重定向，跳轉的url需要參數時，也得傳參，注意：不用加類型 redirect_to("/index/<nid>") subdomain=None 二級域名 ：詳見day118-06 首先你要有一個主域名 動態的二級域名 - 在本地hosts文件中找IP C:\Windows\System32\drivers\etc ios/lenux系統是在/etc/hosts from flask import Flask,render_template,request,redirect,session,url_for app = Flask(__name__) app.config['SERVER_NAME'] = 'bjg.com:5000' @app.route("/index",subdomain='<xxxxx>') def index(xxxxx): return "%s.bjg.com" %(xxxxx,) if __name__ == '__main__': app.run() - 擴展Flask的路由系統，讓他支持正則 from flask import Flask,url_for app = Flask(__name__) # 定義轉換的類 from werkzeug.routing import BaseConverter class RegexConverter(BaseConverter): """ 自定義URL匹配正則表達式 """ def __init__(self, map, regex): super(RegexConverter, self).__init__(map) self.regex = regex def to_python(self, value): """ 路由匹配時，匹配成功后傳遞給視圖函數中參數的值 :param value: :return: """ return int(value) def to_url(self, value): """ 使用url_for反向生成URL時，傳遞的參數經過該方法處理，返回的值用於生成URL中的參數 :param value: :return: """ val = super(RegexConverter, self).to_url(value) return val # 添加到converts中 app.url_map.converters['xxx'] = RegexConverter # 進行使用 @app.route('/index/<xxx("\d+"):nid>',endpoint='xx') def index(nid): url_for('xx',nid=123) return "Index" if __name__ == '__main__': app.run() - 視圖函數 請求： request.files 文件信息 request.values 所有的信息 request.form post請求 request.args get請求 響應： reutrn render_template() reutrn redirect() return "" return jsonify(name='alex',age='18') 返回json格式數據 make_response 每一個return的時候flask都做處理 → make_response(返回值), 可以設置cookie 及session 或者更多的其他內容 eg： response = make_response('xxxxx') response.headers['xxx'] = '123123' return response - CBV、FBV 回顧django的cbv： url(r'^login_cbv/',views.Login.as_view) from django.views import View class Login(View): def get(self,request): return render(request,"login.html") def post(self,request): return HttpResponse("post...") 如果是get請求就走 get方法，如果是post請求就走post方法。 flask的cbv def auth(func): def inner(*args, **kwargs): result = func(*args, **kwargs) return result return inner class IndexView(views.MethodView): methods = ['POST'] #只寫POST時，只有請求時post時才生效，也就是get函數不執行 decorators = [auth,] 為每一個函數都加上auth裝飾器 def get(self): v = url_for('index') print(v) return "GET" def post(self): return "GET" app.add_url_rule('/index', view_func=IndexView.as_view(name='index')) if __name__ == '__main__': app.run() 零碎知識點： root_path 根路徑 to_dict() 把url變成字典 urllib.parse 引入 urlencode 把字典變成url形式 quote 和 unquote 把漢字變成亂碼/把亂碼變成漢字 self.__class__ 找到對象的類 - 模板 1.方法不會自動執行，要加括號，字典也可以用get方法 2.Makeup相當於django的 mark_safe , 或者在前端頁面上 |safe Markup("<input type='text' />") 引出 xss 攻擊 3.自定義的 標簽和 過濾器 @ ，在頁面上寫的區別， 3.1 需要把test傳給前端 def test(a1,a2): return a1+a2 return render_template('index.html',test=test) 使用：{{test(1,19)}} 3.2 所有的模板都可以用 @app.template_global() #加上這個裝飾器以后就不需要傳了，所有的頁面直接就可以使用 def sb(a1, a2): return a1 + a2 + 100 使用：{{sb(1,2)}} 3.3 所有的模板都可以用 @app.template_filter() def db(a1, a2, a3): return a1 + a2 + a3 使用：{{ 1|db(2,3)}} 4.模板的繼承和django一樣的，include也是一樣的 5.hong： {% macro xxxx(name, type='text', value='') %} <input type="{{ type }}" name="{{ name }}" value="{{ value }}"> <input type="{{ type }}" name="{{ name }}" value="{{ value }}"> <input type="{{ type }}" name="{{ name }}" value="{{ value }}"> <input type="{{ type }}" name="{{ name }}" value="{{ value }}"> {% endmacro %} {{ xxxx('n1') }} - 藍圖 - 項目目錄規則化 (把一個py文件分成多個py文件 ) 藍圖：小中型項目：結構 大中型項目：結構 - 特殊裝飾器 app.before_first_request app.before_request app.after_request 必須要有返回值，並且得有一個形參 - session 設置值 session['xx']=1 取值 session.get('xx') 超時時間的設置 session超時時間如何設置？ PERMANENT_SESSION_LIFETIME app.config['SESSION_COOKIE_NAME'] = 'session_lvning' """ '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, #是否實時更新 'PERMANENT_SESSION_LIFETIME': timedelta(days=31) """ session本質是操作的一個字典，在修改session的時候是在內存中操作的，等在return之前 把內存的session字典再返回給你open_session / save_session - flash 同session的原理，區別在於flash取一次就沒了 設置值，flash('xxxx') 取值，get_flashed_messages()
2.上下文管理
什么是上寫文？
上下文個人理解是：程序運行時相關的周圍環境，flask里的上下文是指：在請求剛進來的時候把某個數據或者變量放到一個棧里，等你后面什么時候用就去棧里取就行了。在java 和php中上下文被叫做 HttpRequestContext
a. 創建Local類： { 線程或協程唯一標識: { 'stack':[request],'xxx':[session,] }, 線程或協程唯一標識: { 'stack':[] }, 線程或協程唯一標識: { 'stack':[] }, 線程或協程唯一標識: { 'stack':[] }, } b. 本質 當請求進來之后，將請求相關數據添加到 [request,] 以后使用時：去讀取 請求完成之后，將request從列表中移除。 c. 關系 local= 宋康 = { 線程或協程唯一標識: { 'stack':[] }, 線程或協程唯一標識: { 'stack':[] }, 線程或協程唯一標識: { 'stack':[] }, 線程或協程唯一標識: { 'stack':[] }, } stack=強哥={ push top pop } 存取數據時，要基於stack來做。 d. Flask和Django區別？ - 請求相關數據傳遞方式 - Django： 參數 - Flask： 基於 Local，LocalStack對象 - 問題：多個請求到來會不會混淆 - 單線程 - 多線程 - 協程 解決： from greenlet import getcurrent as get_ident
3. 數據庫連接池
3.1本地線程 -每一個線程來的時候，都分配一個標示，也就是說每個線程都有自己的數據信息，當取值的時候，只取自己線程的數據，這樣實現了線程之間的數據隔離 import threading import time # 本地線程對象 local_values = threading.local() def func(num): """ # 第一個線程進來，本地線程對象會為他創建一個唯一標識 # 第二個線程進來，本地線程對象會為他創建一個唯一標識 { 線程1的唯一標識：{name:1}, 線程2的唯一標識：{name:2}, } """ local_values.name = num # 線程停下來了 time.sleep(2) # local_values.name，去local_values中根據自己的唯一標識作為key，獲取value中name對應的值 print(local_values.name, threading.current_thread().name) for i in range(5): th = threading.Thread(target=func, args=(i,), name='線程%s' % i) th.start() 3.2原來連接數據庫的方式： 1 #缺點：每次請求反復創建數據庫連接，連接數太多 conn = pymysql.connect() cursor = conn.cursor() cursor.execute('select * from tb where id > %s',[5,]) result = cursor.fetchall() cursor.close() conn.close() print(result) 2 # 公用一個連接，多線程有問題，加鎖→缺點，不能支持並發 with LOCK: cursor = CONN.cursor() cursor.execute('select * from tb where id > %s', [5, ]) result = cursor.fetchall() cursor.close() print(result) 3.3數據庫連接池： 原理：設置連接池中最大連接數、默認啟動時連接池中創建的連接數 3.3.1.為每個線程創建一個連接，該線程關閉時，不是真正關閉；本線程再次調用時，還是使用的最開始創建的連接。直到線程終止，數據庫連接才關閉。（本質是用本地線程實現的） ；當很多線程進來時還是會創建很多的連接，所以這種方法也不好。 from DBUtils.PersistentDB import PersistentDB import pymysql import threading POOL = PersistentDB( creator=pymysql, # 使用鏈接數據庫的模塊 maxusage=None, # 一個鏈接最多被重復使用的次數，None表示無限制 setsession=[], # 開始會話前執行的命令列表。如：["set datestyle to ...", "set time zone ..."] ping=0, # ping MySQL服務端，檢查是否服務可用。# 如：0 = None = never, 1 = default = whenever it is requested, 2 = when a cursor is created, 4 = when a query is executed, 7 = always closeable=False, # 如果為False時， conn.close() 實際上被忽略，供下次使用，再線程關閉時，才會自動關閉鏈接。如果為True時， conn.close()則關閉鏈接，那么再次調用pool.connection時就會報錯，因為已經真的關閉了連接（pool.steady_connection()可以獲取一個新的鏈接） threadlocal=None, # 本線程獨享值得對象，用於保存鏈接對象，如果鏈接對象被重置 host='127.0.0.1', port=3306, user='root', password='123', database='pooldb', charset='utf8' ) def func(): conn = POOL.connection() cursor = conn.cursor() cursor.execute('select * from tb1') result = cursor.fetchall() cursor.close() conn.close() # 不是真的關閉，而是假的關閉。 for i in range(10): t = threading.Thread(target=func) t.start() 3.3.2.創建一個連接池(10個連接)，為所有線程提供連接，使用時來進行獲取，使用完畢后，再次放回到連接池。 注意: 連接池中的所有連接都可以被重新使用，原因是因為pymsql.threadsafety的 個數是1 數據庫連接池樣板： xxx.py--------------- from flask import Flask from db import POOL app = Flask(__name__) @app.route('/index') def index(): conn = POOL.connection() cursor = conn.cursor() cursor.execute('select * from tb1') result = cursor.fetchall() conn.close() return '執行成功' if __name__ == '__main__': app.run() db.py--------------- import pymysql from DBUtils.PooledDB import PooledDB, SharedDBConnection POOL = PooledDB( creator=pymysql, # 使用鏈接數據庫的模塊 maxconnections=6, # 連接池允許的最大連接數，0和None表示不限制連接數 mincached=2, # 初始化時，鏈接池中至少創建的空閑的鏈接，0表示不創建 maxcached=5, # 鏈接池中最多閑置的鏈接，0和None不限制 maxshared=3, # 鏈接池中最多共享的鏈接數量，0和None表示全部共享。PS: 無用，因為pymysql和MySQLdb等模塊的 threadsafety都為1，所有值無論設置為多少，_maxcached永遠為0，所以永遠是所有鏈接都共享。 blocking=True, # 連接池中如果沒有可用連接后，是否阻塞等待。True，等待；False，不等待然后報錯 maxusage=None, # 一個鏈接最多被重復使用的次數，None表示無限制 setsession=[], # 開始會話前執行的命令列表。如：["set datestyle to ...", "set time zone ..."] ping=0, # ping MySQL服務端，檢查是否服務可用。# 如：0 = None = never, 1 = default = whenever it is requested, 2 = when a cursor is created, 4 = when a query is executed, 7 = always #ping ： 0 或 4最常用 host='127.0.0.1', port=3306, user='root', password='123', database='pooldb', charset='utf8' ) 如果有三個線程來連接池（最大連接數：9個，初始化創建了5個）拿連接，有三種情況： 1. 他們三個先后順序來， 同一個連接就夠三個線程用了 2. 他們來的順序不定， 有可能需要兩個連接 3. 他們同時來的， 需要三個鏈接
4. 單例模式 ： - 推薦：__new__
- 文件 - 基於類方法 # 單例模式：無法支持多線程情況 """ class Singleton(object): def __init__(self): import time time.sleep(1) @classmethod def instance(cls, *args, **kwargs): if not hasattr(Singleton, "_instance"): Singleton._instance = Singleton(*args, **kwargs) return Singleton._instance import threading def task(arg): obj = Singleton.instance() print(obj) for i in range(10): t = threading.Thread(target=task,args=[i,]) t.start() """ # # 單例模式：支持多線程情況 """ import time import threading class Singleton(object): _instance_lock = threading.Lock() def __init__(self): time.sleep(1) @classmethod def instance(cls, *args, **kwargs): if not hasattr(Singleton, "_instance"): with Singleton._instance_lock: if not hasattr(Singleton, "_instance"): Singleton._instance = Singleton(*args, **kwargs) return Singleton._instance def task(arg): obj = Singleton.instance() print(obj) for i in range(10): t = threading.Thread(target=task,args=[i,]) t.start() time.sleep(20) obj = Singleton.instance() print(obj) """ - 基於__new__方法 引出：new，call，init 方法及區別 """ 1.對象是類創建，創建對象時候類的__init__方法自動執行，對象()執行類的 __call__ 方法 2.類是type創建，創建類時候type的__init__方法自動執行，類() 執行type的 __call__方法(類的__new__方法,類的__init__方法) # 第0步: 執行type的 __init__ 方法【類是type的對象】 class Foo: def __init__(self): pass def __call__(self, *args, **kwargs): pass # 第1步: 執行type的 __call__ 方法 # 1.1 調用 Foo類（是type的對象）的 __new__方法，用於創建對象。 # 1.2 調用 Foo類（是type的對象）的 __init__方法，用於對對象初始化。 obj = Foo() # 第2步：執行Foodef __call__ 方法 obj() """ """ class SingletonType(type): def __init__(self,*args,**kwargs): super(SingletonType,self).__init__(*args,**kwargs) def __call__(cls, *args, **kwargs): obj = cls.__new__(cls,*args, **kwargs) cls.__init__(obj,*args, **kwargs) # Foo.__init__(obj) return obj class Foo(metaclass=SingletonType): def __init__(self,name): self.name = name def __new__(cls, *args, **kwargs): return object.__new__(cls, *args, **kwargs) obj = Foo('name') """ import threading class Singleton(object): _instance_lock = threading.Lock() def __init__(self): pass def __new__(cls, *args, **kwargs): if not hasattr(Singleton, "_instance"): with Singleton._instance_lock: if not hasattr(Singleton, "_instance"): Singleton._instance = object.__new__(cls, *args, **kwargs) return Singleton._instance obj1 = Singleton() obj2 = Singleton() print(obj1,obj2) # <__main__.Singleton object at 0x0000029B7FE127F0> <__main__.Singleton object at 0x0000029B7FE127F0> - 基於metaclass方法 import threading class SingletonType(type): _instance_lock = threading.Lock() def __call__(cls, *args, **kwargs): if not hasattr(cls, "_instance"): with SingletonType._instance_lock: if not hasattr(cls, "_instance"): cls._instance = super(SingletonType,cls).__call__(*args, **kwargs) return cls._instance class Foo(metaclass=SingletonType): def __init__(self,name): self.name = name obj1 = Foo('name') obj2 = Foo('name') print(obj1,obj2) # <__main__.Foo object at 0x0000017B6A612898> <__main__.Foo object at 0x0000017B6A612898> 在哪應用了單例模式： a. stark組件 b. 數據庫連接池
5.Session
5.1- 讀Flask session源碼 5.2- 流程： 當請求第一次進來，生成隨機字符串：sidfnsdfisdfs - 發給用戶cookie - 保存到session字典中 PS: 調用 stack（強哥）將隨機字符串和對應的值放到 local(宋康) 視圖函數使用 session = LocalProxy(partial(_lookup_req_object, 'session')) 請求處理完畢 將session做持久化： - 存到數據 - 存到redis - 存到加密cookie中 5.3- 自定義session： from flask import Flask,request,session app = Flask(__name__) app.secret_key = 'sdfsdfsd' from flask.sessions import SessionInterface,SessionMixin import uuid import json from flask.sessions import SessionInterface from flask.sessions import SessionMixin from itsdangerous import Signer, BadSignature, want_bytes class MySession(dict, SessionMixin): def __init__(self, initial=None, sid=None): self.sid = sid self.initial = initial super(MySession, self).__init__(initial or ()) def __setitem__(self, key, value): super(MySession, self).__setitem__(key, value) def __getitem__(self, item): return super(MySession, self).__getitem__(item) def __delitem__(self, key): super(MySession, self).__delitem__(key) class MySessionInterface(SessionInterface): session_class = MySession container = { # 'asdfasdfasdfas':{'k1':'v1','k2':'v2'} # 'asdfasdfasdfas':"{'k1':'v1','k2':'v2'}" } def __init__(self): pass # import redis # self.redis = redis.Redis() def _generate_sid(self): return str(uuid.uuid4()) def _get_signer(self, app): if not app.secret_key: return None return Signer(app.secret_key, salt='flask-session', key_derivation='hmac') def open_session(self, app, request): """ 程序剛啟動時執行，需要返回一個session對象 """ sid = request.cookies.get(app.session_cookie_name) if not sid: # 生成隨機字符串，並將隨機字符串添加到 session對象中 sid = self._generate_sid() return self.session_class(sid=sid) signer = self._get_signer(app) try: sid_as_bytes = signer.unsign(sid) sid = sid_as_bytes.decode() except BadSignature: sid = self._generate_sid() return self.session_class(sid=sid) # session保存在redis中 # val = self.redis.get(sid) # session保存在內存中 val = self.container.get(sid) if val is not None: try: data = json.loads(val) return self.session_class(data, sid=sid) except: return self.session_class(sid=sid) return self.session_class(sid=sid) def save_session(self, app, session, response): """ 程序結束前執行，可以保存session中所有的值 如： 保存到resit 寫入到用戶cookie """ domain = self.get_cookie_domain(app) path = self.get_cookie_path(app) httponly = self.get_cookie_httponly(app) secure = self.get_cookie_secure(app) expires = self.get_expiration_time(app, session) val = json.dumps(dict(session)) # session保存在redis中 # self.redis.setex(name=session.sid, value=val, time=app.permanent_session_lifetime) # session保存在內存中 self.container.setdefault(session.sid, val) session_id = self._get_signer(app).sign(want_bytes(session.sid)) response.set_cookie(app.session_cookie_name, session_id, expires=expires, httponly=httponly, domain=domain, path=path, secure=secure) app.session_interface = MySessionInterface() # app.session_interface = Foo() # app.session_interface # app.make_null_session() @app.route('/index') def index(): print('網站的所有session',MySessionInterface.container) print(session) session['k1'] = 'v1' session['k2'] = 'v2' del session['k1'] # 在內存中操作字典.... # session['k1'] = 'v1' # session['k2'] = 'v2' # del session['k1'] return "xx" if __name__ == '__main__': app.__call__ app.run() 5.4 最常用 pip3 install flask-session #!/usr/bin/env python # -*- coding:utf-8 - from flask import Flask,current_app,session from flask_session import Session app = Flask(__name__) app.debug = True app.secret_key = 'xxxx' # 為Flask-session組件提供的配置 # import redis # app.config['SESSION_TYPE'] = 'redis' # session類型為redis # app.config['SESSION_REDIS'] = redis.Redis(host='127.0.0.1', port='6379', password='123123') # 用於連接redis的配置 # app.config['SESSION_KEY_PREFIX'] = 'session:' # 保存到session中的值的前綴 # app.config['SESSION_PERMANENT'] = False # 如果設置為True，則關閉瀏覽器session就失效。 # app.config['SESSION_USE_SIGNER'] = False # 是否對發送到瀏覽器上 session:cookie值進行加密 # Session(app) # # import memcache # app.config['SESSION_TYPE'] = 'memcached' # session類型為memcached # app.config['SESSION_PERMANENT'] = True # 如果設置為True，則關閉瀏覽器session就失效。 # app.config['SESSION_USE_SIGNER'] = False # 是否對發送到瀏覽器上session的cookie值進行加密 # app.config['SESSION_KEY_PREFIX'] = 'session:' # 保存到session中的值的前綴 # app.config['SESSION_MEMCACHED'] = memcache.Client(['10.211.55.4:12000']) # Session(app) app.config['SESSION_TYPE'] = 'filesystem' # session類型為filesystem app.config['SESSION_FILE_DIR'] = r'C:\Users\Administrator\PycharmProjects\day121\2.flask-session組件' # session類型為redis app.config['SESSION_FILE_THRESHOLD'] = 500 # 存儲session的個數如果大於這個值時，就要開始進行刪除了 app.config['SESSION_FILE_MODE'] = 384 # 文件權限類型 app.config['SESSION_PERMANENT'] = True # 如果設置為True，則關閉瀏覽器session就失效。 app.config['SESSION_USE_SIGNER'] = False # 是否對發送到瀏覽器上session的cookie值進行加密 app.config['SESSION_KEY_PREFIX'] = 'session:' # 保存到session中的值的前綴 Session(app) @app.route('/index') def index(): session['k1'] = 'v1' return 'xx' if __name__ == '__main__': app.run()
6.blinker 信號
pip3 install blinker 6.1 內置信號 10個信號： 2. request_started = _signals.signal('request-started') # 請求到來前執行 5. request_finished = _signals.signal('request-finished') # 請求結束后執行 3. before_render_template = _signals.signal('before-render-template') # 模板渲染前執行 4. template_rendered = _signals.signal('template-rendered') # 模板渲染后執行 2/3/4/5或不執行 got_request_exception = _signals.signal('got-request-exception') # 請求執行出現異常時執行 6. request_tearing_down = _signals.signal('request-tearing-down') # 請求執行完畢后自動執行（無論成功與否） 7. appcontext_tearing_down = _signals.signal('appcontext-tearing-down')# 請求上下文執行完畢后自動執行（無論成功與否） 1. appcontext_pushed = _signals.signal('appcontext-pushed') # 請求app上下文push時執行 8. appcontext_popped = _signals.signal('appcontext-popped') # 請求上下文pop時執行 message_flashed = _signals.signal('message-flashed') # 調用flask在其中添加數據時，自動觸發 問題： 特殊的裝飾器和信號有什么區別？ - 裝飾器返回值有意義 - 信號用於做什么呢？ - 降低代碼之間的耦合 6.2 自定義信號 ：創建信號→注冊→觸發 from flask import Flask,flash from flask.signals import _signals app = Flask(__name__) wh = _signals.signal('wh') # 定義函數 def luominwen(*args,**kwargs): print('羅姑娘',args,kwargs) # 定義函數 def shaowei(*args,**kwargs): print('少姑娘',args,kwargs) # 將函數注冊到request_started信號中： 添加到這個列表 wh.connect(luominwen) wh.connect(shaowei) @app.route('/index') def index(): # 觸發這個信號：執行注冊到列表中的所有函數 # 發送短信，郵件，微信 wh.send(sender='xxx',a1=123,a2=456) return "xx" if __name__ == '__main__': app.__call__ app.run()
6.3Django內置
Request/response signals
request_started # 請求到來前，自動觸發
request_finished # 請求結束后，自動觸發
got_request_exception # 請求異常后，自動觸發
Model signals
pre_init # django的modal執行其構造方法前，自動觸發
post_init # django的modal執行其構造方法后，自動觸發
pre_save # django的modal對象保存前，自動觸發
post_save # django的modal對象保存后，自動觸發， 它可以判斷是增加的還是修改的
pre_delete # django的modal對象刪除前，自動觸發
post_delete # django的modal對象刪除后，自動觸發
m2m_changed # django的modal中使用m2m字段操作第三張表（add,remove,clear）前后，自動觸發
class_prepared # 程序啟動時，檢測已注冊的app中modal類，對於每一個類，自動觸發
Management signals
pre_migrate # 執行migrate命令前，自動觸發
post_migrate # 執行migrate命令后，自動觸發
Test signals
setting_changed # 使用test測試修改配置文件時，自動觸發
template_rendered # 使用test測試渲染模板時，自動觸發
Database Wrappers
connection_created # 創建數據庫連接時，自動觸發
需求：新浪面試題，數據庫12張表，每張表創建一條數據時，記錄一條日志。
答案： 可以重寫 django里面orm操作的save 方法，或者 使用信號
7. wtforms
- wtforms的源碼
- 使用
metaclass的另外一種方式： class MyType(type): def __init__(self,*args,**kwargs): print('xxxx') super(MyType,self).__init__(*args,**kwargs) def __call__(cls, *args, **kwargs): obj = cls.__new__(cls,*args, **kwargs) cls.__init__(obj,*args, **kwargs) # Foo.__init__(obj) return obj def with_metaclass(base): return MyType("MyType",(base,),{}) class Foo(with_metaclass(object)): def __init__(self,name): self.name = name #打印結果： xxxx xxxx
8. flask的簡單 使用：
- 藍圖 - 配置 from_objects - flask-session - 記錄：請求到來寫日志 - wtforms - DBUtils - 單例模式 PS: 連接數據庫，對用戶表進行：登錄和注冊
9. 導出程序內應用的所有模塊
# 獲取環境中所有安裝的模塊 pip3 freeze #在終端查看 pip3 freeze > requirements.txt #寫入requirements.txt文件中 # pip3 install pipreqs # 獲取當前所在程序目錄中涉及到的所有模塊，並自動生成 requirements.txt 且寫入內容。 pipreqs ./ 以后別人給你程序： requirements.txt 進入程序目錄： pip3 install -r requirements.txt