' 在烧瓶中(SQLAlchemy,Flask-RBAC),python,flask,sqlalchemy"/>

首页 > 解决方案 > 如何修复'NotImplementedError:' 在烧瓶中(SQLAlchemy,Flask-RBAC)

问题描述

每当我尝试登录或注册(在我的数据库中查找的两个操作)时,我都会从后端收到一个奇怪的“NotImplementedError:内置函数 getitem”错误,并返回 500。

该代码今天早些时候可以工作,我一生都无法找出破坏它的变化。我的数据库有一个简单的结构:用户和角色以及它们之间的关联。我可以使用这样的东西将事情提交到我的数据库中:

app = create_app()
app.app_context().push()
with app.app_context():
    db.create_all()

user = User(name="Matt", email="matt@gmail.com", password=generate_password_hash("ligma2008", method='sha256'))
user.add_role("admin") # (role previously created)
db.session.add(user)
db.session.commit()

但是后来我无法登录该用户(或注册一个新用户)。以下是相关代码:

数据库.py:

from flask_sqlalchemy import SQLAlchemy

db = SQLAlchemy()

模型.py:

from flask_rbac import RoleMixin, UserMixin
from .database import db
from .login_manager import login_manager

users_roles = db.Table(
    'users_roles',
    db.Column('user_id', db.Integer, db.ForeignKey('user.id')),
    db.Column('role_id', db.Integer, db.ForeignKey('role.id'))
)

@login_manager.user_loader
def load_user(user_id):
    return User.query.get(user_id)

class User(db.Model, UserMixin):
    id = db.Column(db.Integer, primary_key=True)
    email = db.Column(db.String(100), unique=True)
    password = db.Column(db.String(100))
    name = db.Column(db.String(1000))

    # Other columns
    roles = db.relationship(
        'Role',
        secondary=users_roles,
        backref=db.backref('roles', lazy='dynamic')
    )

    def add_role(self, role):
        self.roles.append(role)

    def add_roles(self, roles):
        for role in roles:
            self.add_role(role)

    def get_roles(self):
        for role in self.roles:
            yield role

roles_parents = db.Table(
    'roles_parents',
    db.Column('role_id', db.Integer, db.ForeignKey('role.id')),
    db.Column('parent_id', db.Integer, db.ForeignKey('role.id'))
)

class Role(db.Model, RoleMixin):
    id = db.Column(db.Integer, primary_key=True)
    name = db.Column(db.String(20))

    parents = db.relationship(
        'Role',
        secondary=roles_parents,
        primaryjoin=(id == roles_parents.c.role_id),
        secondaryjoin=(id == roles_parents.c.parent_id),
        backref=db.backref('children', lazy='dynamic')
    )

    def __init__(self, name):
        RoleMixin.__init__(self)
        self.name = name

    def add_parent(self, parent):
        self.parents.append(parent)

    def add_parents(self, *parents):
        for parent in parents:
            self.add_parent(parent)

    @staticmethod
    def get_by_name(name):
        return Role.query.filter_by(name=name).first()

登录功能:

from werkzeug.security import generate_password_hash, check_password_hash
from flask_login import login_user, logout_user
from .database import db
from .models import User

@auth.route('/login', methods=['POST'])
def login():
    name = request.json["name"]
    password = request.json["password"]
    user = User.query.filter_by(name=name).first()

    # check if the user actually exists
    # take the user-supplied password, hash it, and compare it to the hashed password in the database
    if not user or not check_password_hash(user.password, password):
        flash('Please check your login details and try again.')
        return "/login" # if the user doesn't exist or password is wrong, reload the page

    # if the above check passes, then we know the user has the right credentials
    login_user(user)
    return "/home"

每当我发布到我的登录功能时,后端都会返回 500 以及我在顶部提到的错误消息。这可能是项目其他地方的一些奇怪的细节,因为就像我说的那样,它早些时候使用非常相似的代码工作......

编辑:在 create_app() 我为 db 初始化应用程序:“db.init_app(app)”完整错误回溯:

127.0.0.1 - - [03/Apr/2021 13:46:49] "OPTIONS /login HTTP/1.1" 500 -
Traceback (most recent call last):
  File "D:\Kami\AppData_B\Roaming\Python\Python38\Lib\site-packages\flask\app.py", line 2464, in __call__
    return self.wsgi_app(environ, start_response)
  File "D:\Kami\AppData_B\Roaming\Python\Python38\Lib\site-packages\flask\app.py", line 2450, in wsgi_app
    response = self.handle_exception(e)
  File "D:\Kami\AppData_B\Roaming\Python\Python38\Lib\site-packages\flask_cors\extension.py", line 165, in wrapped_function
    return cors_after_request(app.make_response(f(*args, **kwargs)))
  File "D:\Kami\AppData_B\Roaming\Python\Python38\Lib\site-packages\flask\app.py", line 1867, in handle_exception
    reraise(exc_type, exc_value, tb)
  File "D:\Kami\AppData_B\Roaming\Python\Python38\Lib\site-packages\flask\_compat.py", line 39, in reraise
    raise value
  File "D:\Kami\AppData_B\Roaming\Python\Python38\Lib\site-packages\flask\app.py", line 2447, in wsgi_app
    response = self.full_dispatch_request()
  File "D:\Kami\AppData_B\Roaming\Python\Python38\Lib\site-packages\flask\app.py", line 1945, in full_dispatch_request
    self.try_trigger_before_first_request_functions()
  File "D:\Kami\AppData_B\Roaming\Python\Python38\Lib\site-packages\flask\app.py", line 1993, in try_trigger_before_first_req
    func()
  File "D:\Kami\AppData_B\Roaming\Python\Python38\Lib\site-packages\flask_rbac\__init__.py", line 432, in _setup_acl
    all_roles = {x.get_name() if not isinstance(x, str)
  File "D:\Kami\AppData_B\Roaming\Python\Python38\Lib\site-packages\flask_rbac\__init__.py", line 432, in <setcomp>
    all_roles = {x.get_name() if not isinstance(x, str)
  File "D:\Kami\AppData_B\Roaming\Python\Python38\Lib\site-packages\sqlalchemy\sql\operators.py", line 434, in __getitem__
    return self.operate(getitem, index)
  File "D:\Kami\AppData_B\Roaming\Python\Python38\Lib\site-packages\sqlalchemy\orm\attributes.py", line 289, in operate
    return op(self.comparator, *other, **kwargs)
  File "D:\Kami\AppData_B\Roaming\Python\Python38\Lib\site-packages\sqlalchemy\sql\operators.py", line 434, in __getitem__
    return self.operate(getitem, index)
  File "D:\Kami\AppData_B\Roaming\Python\Python38\Lib\site-packages\sqlalchemy\sql\operators.py", line 225, in operate
    raise NotImplementedError(str(op))
NotImplementedError: <built-in function getitem>

标签: pythonflasksqlalchemy

解决方案


答案是使用 Flask-User 而不是 Flask-RBAC 作为角色,因为后者很糟糕,我无法让它工作。

from flask_user import UserManager

@main.route("/AdminSettings", methods=["GET", "POST"])
@roles_required('administrateur')
def admin_only():
    return "COOL!"

并将用户和角色添加到您的数据库中,请执行以下操作:

user_a = User(name="George", email="goergy02@gmail.com", password=generate_password_hash("YEET_SKEET_REPEAT", method='sha256'))
user_a.roles.append(Role(name='administrateur'))
user_b = User(name="Matt", email="mattheoxxx@gmail.com", password=generate_password_hash("ligma2008", method='sha256'))
user_b.roles.append(Role(name='prep_clients_residentiels'))
user_c = User(name="Ricky", email="rickowacko@gmail.com", password=generate_password_hash("sheesh_kebab", method='sha256'))
user_c.roles.append(Role(name='prep_clients_affaire'))

db.session.add(user_a)
db.session.add(user_b)
db.session.add(user_c)

db.session.commit()

推荐阅读