1. 程式人生 > >Flask進階(路由檢視,session,藍圖,上下文管理)

Flask進階(路由檢視,session,藍圖,上下文管理)

1. 路由+檢視
    a. 路由設定的兩種方式:
        @app.route('/xxx')
            def index():
                return "index"

        
        def index():
            return "index"
        app.add_url_rule("/xxx",None,index)
        
        注意事項:
            - 不用讓endpoint重名
            - 如果重名函式也一定要相同。
    b. 引數
        rule,                       URL規則
        view_func,                  檢視函式名稱
        endpoint=None,              名稱,用於反向生成URL,即: url_for('名稱')
        methods=None,               允許的請求方式,如:["GET","POST"]
        strict_slashes=None,        對URL最後的 / 符號是否嚴格要求,
        redirect_to=None,           重定向到指定地址

        defaults=None,              預設值,當URL中無引數,函式需要引數時,使用defaults={'k':'v'}為函式提供引數
        subdomain=None,             子域名訪問
    c. CBV
        import functools
        from flask import Flask,views
        app = Flask(__name__)


        def wrapper(func):
            @functools.wraps(func)
            def inner(*args,**kwargs):
                return func(*args,**kwargs)

            return inner



        class UserView(views.MethodView):
            methods = ['GET']
            decorators = [wrapper,]

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

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

        app.add_url_rule('/user',None,UserView.as_view('uuuu'))

        if __name__ == '__main__':
            app.run()
    d. 自定義正則
        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

        # 步驟二:新增到轉換器
        app.url_map.converters['reg'] = RegexConverter

        """
        1. 使用者傳送請求
        2. flask內部進行正則匹配
        3. 呼叫to_python(正則匹配的結果)方法
        4. to_python方法的返回值會交給檢視函式的引數

        """

        # 步驟三:使用自定義正則
        @app.route('/index/<reg("\d+"):nid>')
        def index(nid):
            print(nid,type(nid))

            print(url_for('index',nid=987))
            return "index"

        if __name__ == '__main__':
            app.run()
    
2. session實現原理(原始碼)
    - 請求到來之後wsgi會觸發__call__方法,由__call__方法再次呼叫wsgi_app方法
        - 在wsgi_app方法中:
            - 首先將 請求相關+空session 封裝到一個RequestContext物件中,即:ctx。
            - 將ctx交給LocalStack物件,再由LocalStack將ctx新增到Local中,Local結構:
                __storage__ = {
                    1231:{stack:[ctx,] }
                }
            - 根據請求中的cookie中提取名稱為sessionid對應的值,對cookie進行加密+反序列化,再次賦值給ctx中的session
            
            -> 檢視函式
            
            - 把session中的資料再次寫入到cookie中。
            - 將ctx刪除
        - 結果返回給使用者瀏覽器
        - 斷開socket連線
    
3. 藍圖
    目標:給開發者提供目錄結構
    
    其他:
        - 自定義模板、靜態檔案
        - 某一類url新增字首
        - 給一類url新增before_request
    
4. threading.local【和flask無任何關係】
    作用:為每個執行緒建立一個獨立的空間,使得執行緒對自己的空間中的資料進行操作(資料隔離)。
        import threading
        from threading import local
        import time

        obj = local()


        def task(i):
            obj.xxxxx = i
            time.sleep(2)
            print(obj.xxxxx,i)

        for i in range(10):
            t = threading.Thread(target=task,args=(i,))
            t.start()

    問題:
        - 如何獲取一個執行緒的唯一標記? threading.get_ident()
        - 根據字典自定義一個類似於threading.local功能?
            import time
            import threading

            DIC = {}

            def task(i):
                ident = threading.get_ident()
                if ident in DIC:
                    DIC[ident]['xxxxx'] = i
                else:
                    DIC[ident] = {'xxxxx':i }
                time.sleep(2)

                print(DIC[ident]['xxxxx'],i)

            for i in range(10):
                t = threading.Thread(target=task,args=(i,))
                t.start()
            
        - 根據字典自定義一個為每個協程開闢空間進行存取資料。
        
            import time
            import threading
            import greenlet

            DIC = {}

            def task(i):
                
                # ident = threading.get_ident()
                ident = greenlet.getcurrent()
                if ident in DIC:
                    DIC[ident]['xxxxx'] = i
                else:
                    DIC[ident] = {'xxxxx':i }
                time.sleep(2)

                print(DIC[ident]['xxxxx'],i)

            for i in range(10):
                t = threading.Thread(target=task,args=(i,))
                t.start()
        
        - 通過getattr/setattr 構造出來 threading.local的加強版(協程)
            import time
            import threading
            try:
                import greenlet
                get_ident =  greenlet.getcurrent
            except Exception as e:
                get_ident = threading.get_ident

            class Local(object):
                DIC = {}

                def __getattr__(self, item):
                    ident = get_ident()
                    if ident in self.DIC:
                        return self.DIC[ident].get(item)
                    return None

                def __setattr__(self, key, value):
                    ident = get_ident()
                    if ident in self.DIC:
                        self.DIC[ident][key] = value
                    else:
                        self.DIC[ident] = {key:value}
                    

            obj = Local()

            def task(i):
                obj.xxxxx = i
                time.sleep(2)
                print(obj.xxxxx,i)

            for i in range(10):
                t = threading.Thread(target=task,args=(i,))
                t.start()
        
5. 上下文管理(第一次)
    請求到來時候:
        # ctx = RequestContext(self, environ) # self是app物件,environ請求相關的原始資料
        # ctx.request = Request(environ)
        # ctx.session = None
        
        # 將包含了request/session的ctx物件放到“空調”
            {
                1232:{ctx:ctx物件}
                1231:{ctx:ctx物件}
                1211:{ctx:ctx物件}
                1111:{ctx:ctx物件}
                1261:{ctx:ctx物件}
            }
            
    檢視函式:
        from flask import reuqest,session
        
        request.method
        
        
    請求結束:
        根據當前執行緒的唯一標記,將“空調”上的資料移除。