Sqlalchemy：一对多关系与多对多关系的结合

Question

我有一个具有多对多关系的用户和组表

_usergroup_table = db.Table('usergroup_table', db.metadata,
    db.Column('user_id',  db.Integer, db.ForeignKey('user.id')),
    db.Column('group_id', db.Integer, db.ForeignKey('group.id')))

class User(db.Model):
    """Handles the usernames, passwords and the login status"""
    id = db.Column(db.Integer, primary_key=True)
    name = db.Column(db.String(60), nullable=False, unique=True)

class Group(db.Model):
    """Used for unix-style access control."""
    id = db.Column(db.Integer, primary_key=True)
    name = db.Column(db.String(60), nullable=False)
    users = db.relationship('User', secondary=_usergroup_table,
                            backref='groups')

现在我想向user类添加一个主组。当然，我可以只添加一个group_id列和一个到group类的关系，但这有缺点。打电话时我想把所有的人都叫来用户组，包括主配组。主组应始终是组关系的一部分。在

编辑：

似乎该走的路是关联对象

^{pr2}$

我可以用AssociationProxy简化这个过程，但是如何强制每个用户只使用一个主组呢？在

Answer 1

使用GroupMemberships模型来保存关联，而不是使用\u usergroup_table？一个用户可以通过组成员身份拥有多个组，而组成员身份可以包含其他属性，例如给定组是否是关联用户的主组。在

编辑

为了强制限制每个用户只能有一个主组，我将在用户模型中使用验证，这样，任何试图分配多于（或少于）一个主组的尝试都会在保存记录时导致错误。我不知道完全依靠数据库的完整性系统可以达到同样的结果。对验证检查进行编码有很多种方法，documentation显示了一种使用validates（）修饰符的好方法。在

Answer 2

与“boolean flag”方法相比，您最初想到的group_id方法有几个优点。在

首先，它很自然地受到限制，因此每个用户只有一个主组。另一方面，装载user.primary_组意味着ORM可以通过它的主键来标识这个相关的行，并且可以在标识映射中本地查找它，或者发出一个简单的按主键选择，而不是发出一个查询，其中包含一个很难索引的WHERE子句，其中包含一个布尔值。另一个问题是，不需要进入association对象模式，它简化了关联表的使用，并允许SQLAlchemy更有效地处理来自/到该表的加载和更新。在

下面我们使用事件，包括捕获“remove”事件的@validates的新版本（从0.7.7开始），以确保对用户组以及User.primary_组保持同步。（如果在旧版本的0.7上，可以使用属性“remove”事件或AttributeExtension.remove“如果您仍然使用0.6或更早版本，则使用扩展方法）。如果您希望在数据库级别强制执行此操作，则可以使用触发器来验证所需的完整性：

from sqlalchemy import *
from sqlalchemy.orm import *
from sqlalchemy.ext.declarative import declarative_base

Base= declarative_base()

_usergroup_table = Table('usergroup_table', Base.metadata,
    Column('user_id',  Integer, ForeignKey('user.id')),
    Column('group_id', Integer, ForeignKey('group.id')))

class User(Base):
    __tablename__ = 'user'
    id = Column(Integer, primary_key=True)
    name = Column(String(60), nullable=False, unique=True)
    group_id = Column(Integer, ForeignKey('group.id'), nullable=False)
    primary_group = relationship("Group")

    @validates('primary_group')
    def _add_pg(self, key, target):
        self.groups.add(target)
        return target

    @validates('groups', include_removes=True)
    def _modify_groups(self, key, target, is_remove):
        if is_remove and target is self.primary_group:
            del self.primary_group
        return target

class Group(Base):
    __tablename__ = 'group'
    id = Column(Integer, primary_key=True)
    name = Column(String(60), nullable=False)
    users = relationship('User', secondary=_usergroup_table,
                            backref=backref('groups', collection_class=set))

e = create_engine("sqlite://", echo=True)
Base.metadata.create_all(e)

s = Session(e)

g1, g2, g3 = Group(name='g1'), Group(name='g2'), Group(name='g3')
u1 = User(name='u1', primary_group=g1)

u1.groups.update([g2, g3])

s.add_all([
    g1, g2, g3, u1
])
s.commit()

u1.groups.remove(g1)
assert u1.primary_group is None
u1.primary_group = g2
s.commit()