4

我正在尝试有条件地应用装饰器(Flask-HTTPAuth 的login_required)。如果 sky_is_blue == True,我想应用装饰器,如果 False,则不应用。

这需要随叫随到,因为它可能会在应用程序的生命周期内发生变化(实际上实际上并没有那么多,但绝对是出于单元测试的目的,无论如何我都很好奇原因)。

所以我将装饰器包裹在装饰器中。

在 False 情况下行为符合预期(不应用装饰器),但在 True 情况下应用装饰器时遇到问题。我不确定这是我做错了什么,还是与 Flask-HTTPAuth 的奇怪交互。

以下脚本演示了两个单元测试的问题。test_sky_not_blue 通过,但 test_sky_blue 失败并出现堆栈跟踪。

from flask import Flask
from flask.ext.httpauth import HTTPBasicAuth
from functools import update_wrapper, wraps
from flask.ext.testing import TestCase
import unittest


app = Flask(__name__)
app.config['TESTING'] = True

sky_is_blue = True
auth = HTTPBasicAuth()


class ConditionalAuth(object):
    def __init__(self, decorator):
        print("ini with {}".format(decorator.__name__))
        self.decorator = decorator
        update_wrapper(self, decorator)

    def __call__(self, func):
        print("__call__: ".format(func.__name__))

        @wraps(func)
        def wrapped(*args, **kwargs):
            print("Wrapped call, function {}".format(func.__name__))
            if sky_is_blue:
                rv = self.decorator(func(*args, **kwargs))
                return rv
            else:
                rv = func(*args, **kwargs)
                return rv
        return wrapped


@app.route('/')
@ConditionalAuth(auth.login_required)
def index():
    """
    Get a token
    """
    return "OK"


class TestSky(TestCase):
    def create_app(self):
        return app

    def test_sky_blue(self):
        global sky_is_blue
        sky_is_blue = True
        response = self.client.get('/')
        self.assert200(response)

    def test_sky_not_blue(self):
        global sky_is_blue
        sky_is_blue = False
        response = self.client.get('/')
        self.assert200(response)


def suite():
    return unittest.makeSuite(TestSky)

if __name__ == '__main__':
    unittest.main(defaultTest='suite')

我得到的完整堆栈跟踪是:

Traceback (most recent call last):
  File "test.py", line 64, in test_sky_blue
    response = self.client.get('/')
  File "/usr/local/lib/python2.7/site-packages/werkzeug/test.py", line 778, in get
    return self.open(*args, **kw)
  File "/usr/local/lib/python2.7/site-packages/flask/testing.py", line 108, in open
    follow_redirects=follow_redirects)
  File "/usr/local/lib/python2.7/site-packages/werkzeug/test.py", line 751, in open
    response = self.run_wsgi_app(environ, buffered=buffered)
  File "/usr/local/lib/python2.7/site-packages/werkzeug/test.py", line 668, in run_wsgi_app
    rv = run_wsgi_app(self.application, environ, buffered=buffered)
  File "/usr/local/lib/python2.7/site-packages/werkzeug/test.py", line 871, in run_wsgi_app
    app_rv = app(environ, start_response)
  File "/usr/local/lib/python2.7/site-packages/flask/app.py", line 1836, in __call__
    return self.wsgi_app(environ, start_response)
  File "/usr/local/lib/python2.7/site-packages/flask/app.py", line 1820, in wsgi_app
    response = self.make_response(self.handle_exception(e))
  File "/usr/local/lib/python2.7/site-packages/flask/app.py", line 1403, in handle_exception
    reraise(exc_type, exc_value, tb)
  File "/usr/local/lib/python2.7/site-packages/flask/app.py", line 1817, in wsgi_app
    response = self.full_dispatch_request()
  File "/usr/local/lib/python2.7/site-packages/flask/app.py", line 1477, in full_dispatch_request
    rv = self.handle_user_exception(e)
  File "/usr/local/lib/python2.7/site-packages/flask/app.py", line 1381, in handle_user_exception
    reraise(exc_type, exc_value, tb)
  File "/usr/local/lib/python2.7/site-packages/flask/app.py", line 1475, in full_dispatch_request
    rv = self.dispatch_request()
  File "/usr/local/lib/python2.7/site-packages/flask/app.py", line 1461, in dispatch_request
    return self.view_functions[rule.endpoint](**req.view_args)
  File "test.py", line 40, in wrapped
    rv = self.decorator(func(*args, **kwargs))
  File "/usr/local/lib/python2.7/site-packages/flask_httpauth.py", line 48, in login_required
    @wraps(f)
  File "/usr/local/Cellar/python/2.7.11/Frameworks/Python.framework/Versions/2.7/lib/python2.7/functools.py", line 33, in update_wrapper
    setattr(wrapper, attr, getattr(wrapped, attr))
AttributeError: 'str' object has no attribute '__module__'

使用 Python 2.7.11、Flask-HTTPAuth==2.7.1、Flask==0.10.1 进行测试,任何见解都将不胜感激。

4

4 回答 4

4

有趣的是,提出一个问题在帮助一个人解决问题方面是多么有效。

问题是装饰器调用中的括号:

rv = self.decorator(func(*args, **kwargs))

将其更改为以下修复它:

rv = self.decorator(func)(*args, **kwargs)

装饰器需要返回一个函数,但是通过将参数直接传递给 func() 我没有给它机会这样做。

我认为,将其分解为单独的调用会使这一点更清楚:

decorated_function = self.decorator(func)
return decorated_function(*args, **kwargs))
于 2016-02-13T13:50:50.757 回答
2

有条件地打开/关闭授权似乎也使用装饰器的optional关键字参数提供了开箱即用的功能。auth.login_required

API 文档

可选optional参数可以设置为 True 以允许路由在请求中不包含身份验证时也执行,在这种情况下 auth.current_user() 将设置为 None。例子:

@auth.login_required(optional=True)
def private_page():
    user = auth.current_user()
    return "Hello {}!".format(user.name if user is not None else 'anonymous')
于 2020-06-04T19:43:56.580 回答
1

有趣的问题。请注意,如果您想要的只是选择性地绕过身份验证逻辑,那么有一种更简单的方法可以做到这一点,而无需使用新的装饰器。只需将绕过逻辑合并到您的verify_password回调中:

@auth.verify_password
def verify(username, password):
    if not sky_is_blue:
        return True  # let the request through, no questions asked!
    # your authentication logic here
    return False  # this will trigger a 401 response

现在您可以login_required像往常一样应用装饰器,并且身份验证将在任何时候成功sky_is_blue == False

@app.route('/')
@auth.login_required
def index():
    """
    Get a token
    """
    return "OK"

希望这可以帮助!

于 2016-02-13T20:07:37.387 回答
0

如果您需要在所有路由上应用条件身份验证检查而不在所有路由上定义login_required包装器,这是一个解决方案。只需使用before_request钩子:

@app.before_request
def conditional_auth_check():
    if your_condition:
        @auth.login_required
        def _check_login():
            return None

        return _check_login()

login_required不一定需要直接包装路由。

于 2018-01-04T08:47:11.937 回答