在 SQLAlchemy 中，如何防止在一对多关系中重新分配子项？答案

【问题标题】：In SQLAlchemy, how can I prevent the reassignment of a child in a one-to-many relationship?在 SQLAlchemy 中，如何防止在一对多关系中重新分配子项？
【发布时间】：2023-01-24 23:42:10
【问题描述】：

如果成员已经属于不同的组，我如何动态地阻止 SQLAlchemy 将 many(Member) 重新分配给不同的 one(Group)？

关系的语义如下：

Group 可能有零或更多Members
Member可能属于零或者一Groups
一旦分配给Group，Member就不能分配给不同的Group，尝试这样做应该是空操作

示例代码：

import sqlalchemy as sa
import sqlalchemy.orm as orm
from sqlalchemy import event


class Group:
    # Columns
    id = sa.Column(sa.Integer(), primary_key=True)

    # Relationships
    members = orm.relationship('Member', back_populates='group')


class Member:
    # Columns
    id = sa.Column(sa.Integer(), primary_key=True)

    # Relationships
    group_id = sa.Column(sa.Integer(), sa.ForeignKey('group.id'), nullable=True)
    group = orm.relationship('Group', back_populates='members')


@event.listens_for(Group.members, 'append', retval=True)
def _append_member(target, value, initiator):
    if value.mro is not None:
        msg = f'Warning: Member {value.id} already belongs to Group {value.group.id}. Cannot reassign'
        print(msg)

        # do something here to prevent the append
        return None

用法示例：

list_of_lists = get_or_create_sorted_members()

for member_list in list_of_lists:
    group = Group()
    group.members.extend(member_list)

这行得通，期望我无法弄清楚从事件处理程序返回什么以便向 SQLAlchemy 发出不应发生追加的信号。返回 None 会在下一个 session.flush() 期间产生错误。返回任何 orm.interfaces.EXT_XXX 常量都会立即产生错误。

我可以引发异常，但这会阻止后续调用通过，并且如果我通过 Group.members.extend(list_of_members) 添加关系，则没有机会捕获异常以允许其他分配继续。

我可以将直接关系替换为具有唯一约束 member.id 的辅助链接表，但这对于零/一对多来说似乎有点过分了。

我几乎无法控制 get_or_create_sorted_members() 的行为。

【问题讨论】：

不确定我是否正确理解了这一点，但是是否会向外键添加唯一约束并捕获将成员插入另一个组时发生的异常对你有用吗？
@python_user 谢谢。在回答您关于为什么捕获异常不起作用的问题时，我尝试重新组织一下，它确实有效！太棒了！至于你的另一点，向外键添加唯一约束会将关系变成一对一，因为没有两个孩子（本例中的成员）可以有相同的父（组）。
这是有道理的，我现在明白你的问题了，如果你没有得到任何答案，我希望你能在某个时候自己回答这个问题

标签： python sqlalchemy

【解决方案1】：

在事件回调函数中引发异常，然后在调用者中捕获它。如果组织得当，它不会妨碍后续迭代。

# Listener
# note that `retval=Ture` is no longer used, and the function does not return any data

@event.listens_for(Group.members, 'append')
def _append_member(target, value, initiator):
    if value.group_id is not None:
        raise ValueError(f'Member {value.id} already belongs to {value.group_id}')

# caller

list_of_lists = get_or_create_sorted_members()

for member_list in list_of_lists:
    group = Group()
    try:
        group.members.extend(member_list)
    except ValueError as e:
        print(f'Warning: {e}')

使用上面的代码，如果member_list 中的 5 个项目中的 2 个已经分配给一个组，则会打印警告，但其他项目将继续分配。

【讨论】：