1. 程式人生 > >利用functools模組的wraps裝飾器實現Flask的登入驗證

利用functools模組的wraps裝飾器實現Flask的登入驗證

首先看一段官方對functools.wraps的功能描述:

This is a convenience function for invoking update_wrapper() as a function decorator when defining a wrapper function. It is equivalent to partial(update_wrapper, wrapped=wrapped,assigned=assigned, updated=updated).

Without the use of this decorator factory, the name of the example function would have been 'wrapper'

, and the docstring of the original example() would have been lost.

如果不使用functools的wraps函式,則被包裹的函式名稱和doc文字內容都會被替換為包裹者函式的相關內容了。

下面的例子就是演示了該情形。例子轉自裝飾器之functools模組的wraps的用途

首先我們先寫一個裝飾器



# 探索functools模組wraps裝飾器的用途
from functools import wraps


def trace(func):
    """ 裝飾器 """

    # @wraps(func)
    def callf(*args, **kwargs):
        """ A wrapper function """
        print("Calling function:{}".format(func.__name__))  # Calling function:foo
        res = func(*args, **kwargs)
        print("Return value:{}".format(res))  # Return value:9
        return res

    return callf


@trace
def foo(x):
    """ 返回給定數字的平方 """
    return x * x


if __name__ == '__main__':
    print(foo(3))  # 9
    print(foo.__doc__)
    help(foo)
    print(foo.__name__)
    # print(foo.__globals__)
    t = trace(foo)
    print(t)
列印結果:
Calling function:foo
Return value:9
9
 A wrapper function 
Help on function callf in module __main__:

callf(*args, **kwargs)
    A wrapper function

callf
<function trace.<locals>.callf at 0x0000022F744D8730>

 

上面的裝飾器例子等價於:trace(foo(3)),只是在使用裝飾器時,我們不用再手動呼叫裝飾器函式;

如果把這段程式碼提供給其他人呼叫, 他可能會想看下foo函式的幫助資訊時:

1

2

3

4

5

6

>>>from xxx import foo

>>>help(foo)    # print(foo__doc__)

Help on function callf in module __main__:

 

callf(*args, **kwargs)

    A wrapper function

這裡,他可能會感到迷惑,繼續敲:

1

2

>>> print(foo.__name__)

callf

最後, 他可能會看原始碼,找問題原因,我們知道Python中的物件都是"第一類"的,所以,trace函式會返回一個callf閉包函式,連帶callf的上下文環境一併返回,所以,可以解釋我們執行help(foo)的到結果了

那麼,我們如果才能得到我們想要的foo的幫助資訊呢,這裡就要用到了functools的wraps了。

# 探索functools模組wraps裝飾器的用途
from functools import wraps


def trace(func):
    """ 裝飾器 """

    @wraps(func)
    def callf(*args, **kwargs):
        """ A wrapper function """
        print("Calling function:{}".format(func.__name__))  # Calling function:foo
        res = func(*args, **kwargs)
        print("Return value:{}".format(res))  # Return value:9
        return res

    return callf


@trace
def foo(x):
    """ 返回給定數字的平方 """
    return x * x


if __name__ == '__main__':
    print(foo(3))  # 9
    print(foo.__doc__)
    help(foo)
    print(foo.__name__)
    # print(foo.__globals__)
    t = trace(foo)
    print(t)

至於wraps的原理,通過下面部分原始碼,可以自行研究,在此就不予展開擴充套件了

 

# 有關wraps的原始碼,有興趣的可以自行研究下


WRAPPER_ASSIGNMENTS = ('__module__', '__name__', '__qualname__', '__doc__',
                       '__annotations__')
WRAPPER_UPDATES = ('__dict__',)
def update_wrapper(wrapper,
                   wrapped,
                   assigned = WRAPPER_ASSIGNMENTS,
                   updated = WRAPPER_UPDATES):
    """Update a wrapper function to look like the wrapped function

       wrapper is the function to be updated
       wrapped is the original function
       assigned is a tuple naming the attributes assigned directly
       from the wrapped function to the wrapper function (defaults to
       functools.WRAPPER_ASSIGNMENTS)
       updated is a tuple naming the attributes of the wrapper that
       are updated with the corresponding attribute from the wrapped
       function (defaults to functools.WRAPPER_UPDATES)
    """
    for attr in assigned:
        try:
            value = getattr(wrapped, attr)
        except AttributeError:
            pass
        else:
            setattr(wrapper, attr, value)
    for attr in updated:
        getattr(wrapper, attr).update(getattr(wrapped, attr, {}))
    # Issue #17482: set __wrapped__ last so we don't inadvertently copy it
    # from the wrapped function when updating __dict__
    wrapper.__wrapped__ = wrapped
    # Return the wrapper so this can be used as a decorator via partial()
    return wrapper

def wraps(wrapped,
          assigned = WRAPPER_ASSIGNMENTS,
          updated = WRAPPER_UPDATES):
    """Decorator factory to apply update_wrapper() to a wrapper function

       Returns a decorator that invokes update_wrapper() with the decorated
       function as the wrapper argument and the arguments to wraps() as the
       remaining arguments. Default arguments are as for update_wrapper().
       This is a convenience function to simplify applying partial() to
       update_wrapper().
    """
    return partial(update_wrapper, wrapped=wrapped,
                   assigned=assigned, updated=updated)








class partial:
    """New function with partial application of the given arguments
    and keywords.
    """

    __slots__ = "func", "args", "keywords", "__dict__", "__weakref__"

    def __new__(*args, **keywords):
        if not args:
            raise TypeError("descriptor '__new__' of partial needs an argument")
        if len(args) < 2:
            raise TypeError("type 'partial' takes at least one argument")
        cls, func, *args = args
        if not callable(func):
            raise TypeError("the first argument must be callable")
        args = tuple(args)

        if hasattr(func, "func"):
            args = func.args + args
            tmpkw = func.keywords.copy()
            tmpkw.update(keywords)
            keywords = tmpkw
            del tmpkw
            func = func.func

        self = super(partial, cls).__new__(cls)

        self.func = func
        self.args = args
        self.keywords = keywords
        return self

    def __call__(*args, **keywords):
        if not args:
            raise TypeError("descriptor '__call__' of partial needs an argument")
        self, *args = args
        newkeywords = self.keywords.copy()
        newkeywords.update(keywords)
        return self.func(*self.args, *args, **newkeywords)

    @recursive_repr()
    def __repr__(self):
        qualname = type(self).__qualname__
        args = [repr(self.func)]
        args.extend(repr(x) for x in self.args)
        args.extend(f"{k}={v!r}" for (k, v) in self.keywords.items())
        if type(self).__module__ == "functools":
            return f"functools.{qualname}({', '.join(args)})"
        return f"{qualname}({', '.join(args)})"

    def __reduce__(self):
        return type(self), (self.func,), (self.func, self.args,
               self.keywords or None, self.__dict__ or None)

    def __setstate__(self, state):
        if not isinstance(state, tuple):
            raise TypeError("argument to __setstate__ must be a tuple")
        if len(state) != 4:
            raise TypeError(f"expected 4 items in state, got {len(state)}")
        func, args, kwds, namespace = state
        if (not callable(func) or not isinstance(args, tuple) or
           (kwds is not None and not isinstance(kwds, dict)) or
           (namespace is not None and not isinstance(namespace, dict))):
            raise TypeError("invalid partial state")

        args = tuple(args) # just in case it's a subclass
        if kwds is None:
            kwds = {}
        elif type(kwds) is not dict: # XXX does it need to be *exactly* dict?
            kwds = dict(kwds)
        if namespace is None:
            namespace = {}

        self.__dict__ = namespace
        self.func = func
        self.args = args
        self.keywords = kwds

只要是知道,wraps是通過partial和update_wrapper來幫我們實現想要的結果的。

實際應用中,例如使用Flask進行開發時,對於一些敏感的頁面是需要登入才能進行操作的,例如部落格頁面的瀏覽(/blog/list),部落格的新增(/blog/add)和部落格的刪除(/blog/del/<int:id>)都應該設計為針對登入使用者的操作。所以,如果一旦在位址列中直接訪問上述受限地址應該予以跳轉到登入頁面,先進行登入再繼續後續的操作。

假設現在我們已經有了針對部落格瀏覽,新增和刪除的路由:

app.route('/blog/add', methods=['GET', 'POST'])
def blog_add():
    ...
    return render_template('art_add.html', ...)

@app.route('/blog/del/<int:id>', methods=['GET'])
def blog_delete(id):
    ...
    return redirect('/art/list', ...)


@app.route('/blog/list', methods=['GET'])
def blog_list():
    return render_template('art_list.html', ...)

那麼在這些檢視函式獲得執行之前,必須要做一次登入驗證,這裡就利用裝飾器進行一次AOP操作:

#裝飾器函式
def login_required(view_func):
    # 5
    @wraps(view_func)     
    def login_req_wrapper(*args,**kwargs):
        if 'user' not in session: # 1
            s = url_for('login',next_url=request.url) # 2
            return redirect(s) # 3
        else:
            return view_func(*args,**kwargs) #4
    return login_req_wrapper

login_required裝飾器函式有如下幾個點需要注意:

  1. 判斷session中是否有user這個鍵。這個操作實際需要一個前提條件就是登入時要在session中新增一個user鍵對應當前登入的使用者名稱稱,登出時將user鍵值對從session中刪除。
  2. 利用url_for構造訪問路徑。login為檢視函式的名稱,根據該檢視函式可以找到路徑/login,後面next_url可以為路徑/login新增一個訪問引數,引數名稱為next_url,引數值為使用者試圖在未登入時訪問的路徑。利用該引數,可以在使用者登入後直接跳轉到之前要訪問的頁面。
  3. 注意,不要遺漏return關鍵字。
  4. 如果session中有user關鍵字意為目前有處於登入狀態的使用者,則直接呼叫路由對應的檢視函式即可。並將檢視函式執行結果利用return返回。
  5. 注意,這裡必須使用@wraps。如果不使用@wraps,則每一個被login_required修飾的檢視函式(view_func)都會有一樣的名字login_req_wrapper,這在Flask中是不允許的!換句話說,Flask中每一個路由都需要有一個名字不同的檢視函式負責響應。

現在將裝飾器login_required新增在需要被驗證的檢視函式上即可:

app.route('/blog/add', methods=['GET', 'POST'])
@login_required
def blog_add():
    ...
    return render_template('art_add.html', ...)

@app.route('/blog/del/<int:id>', methods=['GET'])
@login_required
def blog_delete(id):
    ...
    return redirect('/art/list', ...)


@app.route('/blog/list', methods=['GET'])
@login_required
def blog_list():
    return render_template('art_list.html', ...)

此時,如果需要瀏覽部落格列表,在未登入的情況下訪問地址:埠號/blog/list會跳轉到登入介面:地址:埠號/login?next_url=...