>我正在尝试有条件地应用装饰器(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 进行测试,任何见解将不胜感激。
,布置问题在帮助解决问题方面是多么有效。
问题是装饰器调用中的括号:
rv = self.decorator(func(*args, **kwargs))
将其更改为以下内容可以修复它:
rv = self.decorator(func)(*args, **kwargs)
装饰器需要返回一个函数,但是通过将参数直接传递给func((,我没有给它这样做的机会。
我认为,将其分解为一个单独的调用会更清楚地说明这一点:
decorated_function = self.decorator(func)
return decorated_function(*args, **kwargs))
有条件地打开/关闭授权似乎也是使用 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')
有趣的问题。请注意,如果您只想选择绕过身份验证逻辑,则有一种更简单的方法可以做到这一点,而无需使用新的装饰器。只需将旁路逻辑合并到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"
希望这有帮助!
如果您需要对所有路由应用条件身份验证检查而不在所有路由上定义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
不一定需要直接包装路由。
参考,有一个小要求,即仅在一个路由上应用条件身份验证,而在其他运行状况检查路由上不应用身份验证。这就是我通过扩展@epoc的答案来做到这一点的方式:
@app.before_request
def conditional_auth_check():
"""
Conditional Authentication
1. If called predict
a. with a certain system header, let it pass
b. Else, check basic auth
2. Anything else - let it pass
"""
if request.path == "/predict":
# Check Header Auth here
if request.headers.get("my-header") and os.getenv("system-enabled-var"):
print("Authorizing using headers")
return
# Check Basic Auth here
print("Authorizing using Basic Auth")
@basic_auth.login_required
def _check_login():
return None
return _check_login()
# let all other traffic pass
return
就我而言,system-enabled-var
永远不能由用户启用,只能由某个系统启用。