Flask（SQLAlchemy、Flask-RBAC）登录报getitem NotImplementedError 500错误

故障现象 #

执行登录、注册这类需要查询数据库的操作时，后端抛出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") # admin角色为预先创建的测试数据
db.session.add(user)
db.session.commit()

但写入测试数据后，既无法登录刚创建的Matt账号，也无法完成新用户注册，只要接口触发请求就会报错。

相关业务代码 #

数据库初始化（database.py） #

from flask_sqlalchemy import SQLAlchemy

db = SQLAlchemy()

数据模型（models.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))

    # 用户与角色的多对多关系
    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()

    # 校验用户是否存在、密码是否匹配
    if not user or not check_password_hash(user.password, password):
        flash('Please check your login details and try again.')
        return "/login"

    # 校验通过后执行登录
    login_user(user)
    return "/home"

补充配置说明 #

已在create_app()函数中通过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_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>

故障原因与修复方案 #

• 根因：User模型中定义与Role的多对多关系时，反向引用的命名与flask_rbac内置的RoleMixin属性重名，导致属性被覆盖。
  原本的反向引用db.backref('roles', lazy='dynamic')会给Role类新增一个名为roles的关系属性，这个属性的作用是查询绑定了当前角色的所有用户；但RoleMixin内部本身就维护了名为roles的集合，用来存储角色关联的权限、父角色列表，供flask_rbac的ACL初始化逻辑读取。
  当flask_rbac执行_setup_acl初始化逻辑时，尝试读取Role实例的roles属性获取角色集合，拿到的却是SQLAlchemy的动态查询对象；代码尝试对这个查询对象做索引取值时，就会触发SQLAlchemy的NotImplementedError: <built-in function getitem>错误，这个错误发生在请求进入接口逻辑之前的首次请求初始化阶段，因此表现为接口直接返回500。
• 修复方法：
  1. 把User模型中roles关系的反向引用名称从roles改成符合语义的users（用来表示绑定当前角色的所有用户），修改后的关系定义如下：

     roles = db.relationship(
         'Role',
         secondary=users_roles,
         backref=db.backref('users', lazy='dynamic')
     )
     

  2. 修正Role模型中父子自关联的连接条件，避免在类作用域中直接引用id属性导致的表达式解析错误，将连接条件修改为使用lambda延迟解析表达式，等类完全加载之后再生成连接规则：

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

内容的提问来源于stack exchange，提问作者Xavier Gravel