我正在为我的 Flask 应用程序编写权限系统，但在弄清楚如何在数据库中查找关系时遇到了问题。我的 github 仓库，如果你想查看所有内容。装饰器旨在限制对装饰视图的访问。
def user_has(attribute):
"""
Takes an attribute (a string name of either a role or an ability) and returns the function if the user has that attribute
"""
def wrapper(func):
@wraps(func)
def inner(*args, **kwargs):
attribute_object = Role.query.filter_by(name=attribute).first() or \
Ability.query.filter_by(name=attribute).first()
if attribute_object in current_user.roles or attribute in current_user.roles.abilities.all():
return func(*args, **kwargs)
else:
# Make this do someting way better.
return "You do not have access"
return inner
return wrapper
我正在使用 SQLAlchemy 并将用户、角色和能力存储在数据库中。用户可能有一个或多个角色。角色可能具有一种或多种能力。我想获取传递给装饰器的字符串并检查用户是否具有该角色，或者用户的角色之一是否具有该能力。装饰器不应该关心它是否被角色或能力参数调用。
显然，这种方法 ( current_user.roles.abilities.all()) 无法通过我的关系数据库，因为我试图在这里找到abilities. 我收到一条错误消息：
AttributeError: 'InstrumentedList' object has no attribute 'abilities'
如何将字符串参数与我当前用户从他/她的角色中获得的能力进行比较？
作为参考，我的模型：
user_role_table = db.Table('user_role',
db.Column(
'user_id', db.Integer, db.ForeignKey('user.uid')),
db.Column(
'role_id', db.Integer, db.ForeignKey('role.id'))
)
role_ability_table = db.Table('role_ability',
db.Column(
'role_id', db.Integer, db.ForeignKey('role.id')),
db.Column(
'ability_id', db.Integer, db.ForeignKey('ability.id'))
)
class Role(db.Model):
__tablename__ = 'role'
id = db.Column(db.Integer, primary_key=True)
name = db.Column(db.String(120), unique=True)
abilities = db.relationship(
'Ability', secondary=role_ability_table, backref='roles')
def __init__(self, name):
self.name = name.lower()
def __repr__(self):
return '<Role {}>'.format(self.name)
def __str__(self):
return self.name
class Ability(db.Model):
__tablename__ = 'ability'
id = db.Column(db.Integer, primary_key=True)
name = db.Column(db.String(120), unique=True)
def __init__(self, name):
self.name = name.lower()
def __repr__(self):
return '<Ability {}>'.format(self.name)
def __str__(self):
return self.name
class User(db.Model):
__tablename__ = 'user'
uid = db.Column(db.Integer, primary_key=True)
email = db.Column(db.String(120), unique=True)
pwdhash = db.Column(db.String(100))
roles = db.relationship('Role', secondary=user_role_table, backref='users')
def __init__(self, email, password, roles=None):
self.email = email.lower()
# If only a string is passed for roles, convert it to a list containing
# that string
if roles and isinstance(roles, basestring):
roles = [roles]
# If a sequence is passed for roles (or if roles has been converted to
# a sequence), fetch the corresponding database objects and make a list
# of those.
if roles and is_sequence(roles):
role_list = []
for role in roles:
role_list.appen(Role.query.filter_by(name=role).first())
self.roles = role_list
# Otherwise, assign the default 'user' role. Create that role if it
# doesn't exist.
else:
r = Role.query.filter_by(name='user').first()
if not r:
r = Role('user')
db.session.add(r)
db.session.commit()
self.roles = [r]
self.set_password(password)
def set_password(self, password):
self.pwdhash = generate_password_hash(password)
def check_password(self, password):
return check_password_hash(self.pwdhash, password)
def is_authenticated(self):
return True
def is_active(self):
return True
def is_anonymous(self):
return False
def get_id(self):
return unicode(self.uid)
def __repr__(self):
return '<User {}>'.format(self.email)
def __str__(self):
return self.email
和装饰的视图：
@app.route('/admin', methods=['GET', 'POST'])
@user_has('admin')
def admin():
users = models.User.query.all()
forms = {user.uid: RoleForm(uid=user.uid, roles=[role.id for role in user.roles])
for user in users}
if request.method == "POST":
current_form = forms[int(request.form['uid'])]
if current_form.validate():
u = models.User.query.get(current_form.uid.data)
u.roles = [models.Role.query.get(role)
for role in current_form.roles.data]
db.session.commit()
flash('Roles updated for {}'.format(u))
return render_template('admin.html', users=users, forms=forms)