在SQLAlchemy中查询多对多关系

Question

我有以下几个模型。一个用户可以有很多角色，而一个角色可以拥有很多权限。我现在有点搞不清楚该怎么查询才能得到我想要的结果。

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

role_permission = db.Table(
    'role_permission', 
    db.Column('permission_id', db.Integer, db.ForeignKey('permission.id')),
    db.Column('role_id', db.Integer, db.ForeignKey('role.id')),
    db.UniqueConstraint('permission_id', 'role_id')
)

class Role(Base):
    __tablename__ = 'role'

    id = db.Column(db.Integer, primary_key=True)
    name = db.Column(db.String(100), unique=True, nullable=False)

class Permission(Base):
    __tablename__ = 'permission'

    id = db.Column(db.Integer, primary_key=True)
    name = db.Column(db.String(100), nullable=False)

    roles = db.relation(Role, secondary=role_permission, backref=db.backref('permissions'))

class User(Base, UserMixin):
    __tablename__ = 'user'

    id = db.Column(db.Integer, primary_key=True)
    username = db.Column(db.String(60), unique=True, nullable=False)
    password_hash = db.Column(db.String(80), nullable=False)

    roles = db.relation(Role, secondary=user_role, backref=db.backref('users'))

我想要得到一个（最好是唯一的）用户所拥有的所有权限的列表，但我似乎找不到合适的方法来实现。

我可以通过在用户模型中创建一个生成器来获取这个列表：

def get_all_permissions(self):
    for role in self.roles:
        for perm in role.permissions:
            yield perm

不过我希望能通过一次查询就能完成这个操作。

Answer 1

如果你已经把用户对象连同权限和角色一起加载到内存中，那么你的代码应该能快速处理这些信息，而不需要再去数据库查找。

否则，你可以使用以下查询：

    user_id = 789
    permissions = (db.session.query(Permission)
            .join(Role, Permission.roles)
            .join(User, Role.users)
            .filter(User.id == user_id)
            ).distinct()
    #print(permissions)
    for perm in permissions:
        print(perm)

Answer 2

好吧，如果你只是想获取权限的列表，可以试试这样做：

permissions = session.query(Permission).\
    join(Role).join(User).filter(User.username='MisterX').all()

或者你可以根据需要进行筛选。如果想让权限变得唯一，可以使用分组：

permissions = session.query(Permission.id, Permission.name).join(Role).join(User).\
    filter(User.username='MisterX').group_by(Permission.id).all()

或者，如果可以的话，使用声明式扩展，而不需要特别的查询：

permissions = User.roles.permissions

这样有帮助吗？

Answer 3

可能是因为你没有指定元数据参数，所以你的关联表没有被正确识别。这个脚本对我来说是有效的：

#!/bin/python

from sqlalchemy import Table
from sqlalchemy import Integer, String, ForeignKey, create_engine, Column, PrimaryKeyConstraint
from sqlalchemy.orm import relationship, backref, sessionmaker
from sqlalchemy.ext.declarative import declarative_base

engine = create_engine('sqlite:///:memory:', echo=True)
Base = declarative_base()

user_role = Table(
    'user_role',
    Base.metadata,
    Column('user_id', Integer, ForeignKey('users.id')),
    Column('role_id', Integer, ForeignKey('roles.id')),
    PrimaryKeyConstraint('user_id', 'role_id')
)

role_permission = Table(
    'role_permission', 
    Base.metadata,
    Column('permission_id', Integer, ForeignKey('permissions.id')),
    Column('role_id', Integer, ForeignKey('roles.id')),
    PrimaryKeyConstraint('permission_id', 'role_id')
)

class Role(Base):
    __tablename__ = 'roles'

    id = Column(Integer, primary_key=True)
    name = Column(String(100), unique=True, nullable=False)

class Permission(Base):
    __tablename__ = 'permissions'

    id = Column(Integer, primary_key=True)
    name = Column(String(100), nullable=False)

    roles = relationship("Role", secondary=role_permission, backref=backref('permissions'))

class User(Base):
    __tablename__ = 'users'

    id = Column(Integer, primary_key=True)
    username = Column(String(60), unique=True, nullable=False)
    password_hash = Column(String(80), nullable=False)

    roles = relationship("Role", secondary=user_role, backref=backref('users'))


Base.metadata.create_all(engine)
session = sessionmaker(bind=engine)()

u = User(username="user", password_hash="secret")

r1 = Role(name="Role 1")
session.add(r1)
r2 = Role(name="Role 2")
session.add(r2)

p1 = Permission(name="Permission 1")
p2 = Permission(name="Permission 2")
p3 = Permission(name="Permission 3")

r1.permissions.append(p1)
r1.permissions.append(p2)

r2.permissions.append(p2)
r2.permissions.append(p3)


u.roles.append(r1)
u.roles.append(r2)

session.add(u)

for perm in session.query(Permission).join(Role, Permission.roles).\
    join(User, Role.users).filter(User.username=="user").distict()all():
    print(perm.name)