我以前认为，在很多很多关系中（可能是所有可用的关系）
parent.children.append(child)
和
child.parents.append(parent)
是等价的。但是，我有一个相当复杂的设置，它为我的类的追加和删除方法激活事件监听器。
这可能是一个很长的镜头，但我写了一个自包含的例子来解释我的问题。简而言之，它包含：
Parent，Child和Pet。哪里
Parent与Child和Pet有多对多的关系
Child与Pet有一对多的关系event.listens_for事件
@sa.event.listens_for(Parent.children, 'append')
@sa.event.listens_for(Parent.children, 'remove')
@sa.event.listens_for(Parent.pets, 'remove')unittest的测试设置，有两个测试：
test_child_pet_relationship_on_parents_combined测试来自Child的东西的附加物，如child1.parents.append(test_parent)，其中child1是Child的一个实例，test_parent是Parent的一个实例
test_child_pet_relationship_on_parents_combined_reversed做同样的事情，但反过来，像test_parent.children.append(child1)。logging设置记录每个查询发生的任何事情。我在每个测试中以及第一个事件监听器中设置了一些草率的记录器条目。即，使用A的日志指的是第一个测试，A'指的是第二个测试。基金会
import logging
import sys
import sqlalchemy as sa
from sqlalchemy.ext.declarative import declarative_base
# setup logger
stdout_handler = logging.StreamHandler(sys.stdout)
formatter = logging.Formatter('%(asctime)s:%(filename)s:%(lineno)d\t%(levelname)s\t%(message)s')
stdout_handler.setFormatter(formatter)
logger = logging.getLogger('sqlalchemy.engine')
logger.setLevel(logging.DEBUG)
logger.addHandler(stdout_handler)
Base = declarative_base()
# many to many relationship between parents and children
parents_children_relationship = sa.Table('parents_children_relationship',
Base.metadata,
sa.Column('parent_id', sa.Integer, sa.ForeignKey('parents.id')),
sa.Column('child_id', sa.Integer, sa.ForeignKey('children.id')),
sa.UniqueConstraint('parent_id', 'child_id'))
# many to many relationship between User and Pet
parents_pets_relationship = sa.Table('parents_pets_relationship',
Base.metadata,
sa.Column('parent_id', sa.Integer, sa.ForeignKey('parents.id')),
sa.Column('pet_id', sa.Integer, sa.ForeignKey('pets.id')),
sa.UniqueConstraint('parent_id', 'pet_id'))
class Parent(Base):
__tablename__ = 'parents'
id = sa.Column(sa.Integer, primary_key=True)
name = sa.Column(sa.String(64))
# many to many relationship between parent and children
# my case allows for a children to have many parents. Don't ask.
children = sa.orm.relationship('Child',
secondary=parents_children_relationship,
backref=sa.orm.backref('parents', lazy='dynamic'),
lazy='dynamic')
# many to many relationship between parents and pets
pets = sa.orm.relationship('Pet',
secondary=parents_pets_relationship,
backref=sa.orm.backref('parents', lazy='dynamic'), #
lazy='dynamic')
def __repr__(self):
return '<Parent (name=%r)>' % (self.name)
class Child(Base):
__tablename__ = 'children'
id = sa.Column(sa.Integer, primary_key=True)
name = sa.Column(sa.String(64))
# parents = <backref relationship with User model>
# one to many relationship with pets
pets = sa.orm.relationship('Pet', backref='child', lazy='dynamic')
def __repr__(self):
return '<Child (name=%r)>' % (self.name)
class Pet(Base):
__tablename__ = 'pets'
id = sa.Column(sa.Integer, primary_key=True)
name = sa.Column(sa.String(64))
# child = backref relationship with cities
child_id = sa.Column(sa.Integer, sa.ForeignKey('children.id'), nullable=
True)
# parents = <relationship backref from User>
def __repr__(self):
return '<Pet (name=%r)>' % (self.name)
from sqlalchemy.orm import object_session
@sa.event.listens_for(Parent.children, 'append')
def _on_append_children(parent, child, initiator):
"""
If a new child is appended to the parent, this listener
will also add the pets bound to the child being bound to the parent.
"""
# appends also the pets bound to the child that the
# parent is being appended to
logger.debug(f'**********1. adding the pets of {child} to {parent}***************')
object_session(parent).execute(
"INSERT INTO parents_pets_relationship VALUES "
"(:parent_id, :pet_id)",
[
{"parent_id": parent.id, "pet_id": pet.id}
for pet in child.pets
]
)
logger.debug('**********1. done!***************')
@sa.event.listens_for(Parent.children, 'remove')
def _on_remove_children(parent, child, initiator, *args, **kwargs):
"""
If a child is removed from the parent, this listener
will also remove only remove_single_pet --> <Pet>
"""
object_session(parent).execute(
"DELETE FROM parents_pets_relationship WHERE "
"parent_id=:parent_id AND pet_id=:pet_id",
[
{"parent_id": parent.id, "pet_id": pet.id}
for pet in child.pets
]
)
@sa.event.listens_for(Parent.pets, 'remove')
def _on_remove_pets(parent, pet, initiator, *args, **kwargs):
"""
If a pet is removed from the parent, and the parent also is related
to the child that has access to that pet, then
* removes relationship with the child, and
* keeps relationship with the remaining pets, except the one that was
removed
"""
object_session(parent).execute(
"DELETE FROM parents_children_relationship WHERE "
"parent_id=:parent_id AND child_id=:child_id",
{"parent_id": parent.id, "child_id": pet.child.id}
)
#### test ###
import unittest
from sqlalchemy import create_engine
from sqlalchemy.orm import sessionmaker
class BasicTestModelCase(unittest.TestCase):
def setUp(self):
self.engine = create_engine("sqlite://", echo=False)
Base.metadata.create_all(self.engine)
Session = sessionmaker(bind=self.engine)
self.session = Session()
def tearDown(self):
Base.metadata.drop_all(bind=self.engine)
def test_child_pet_relationship_on_parents_combined(self):
"""
Test that a parent can be hold children and pets that don't
belong necessary to the child, given the behaviour tested in the
previous test.
"""
# create new parent
test_parent = Parent(name='test_parent')
child1 = Child(id=1,
name='FakeChild1')
child2 = Child(id=2,
name='FakeChild2')
pet1 = Pet(id=1,
name='FakePet1',
child_id=1)
pet2 = Pet(id=2,
name='FakePet2',
child_id=2)
pet3 = Pet(id=3,
name='FakePet3',
child_id=1)
self.session.add(test_parent)
self.session.add(child1)
self.session.add(child2)
self.session.add(pet1)
self.session.add(pet2)
self.session.add(pet3)
self.session.commit()
# add parent to the child
logger.debug('************A - add test_parent to child1***************')
child1.parents.append(test_parent)
self.session.add(child1)
self.session.commit()
logger.debug('**********A - done!***************')
# add parent to the child
pet2.parents.append(test_parent)
logger.debug('************B - add test_parent to child1***************')
# persist changes in the db
self.session.add(pet2)
self.session.commit()
logger.debug('**********B - done!***************')
print(test_parent.pets.all())
print(child2.pets.all())
# check that previous relationships are intact
self.assertTrue(child1.pets.all() == [pet1, pet3])
self.assertTrue(child2.pets.all() == [pet2])
# resultant elements should be only child1, its pets and the single Pet
self.assertTrue(test_parent.children.all() == [child1])
self.assertTrue(test_parent.pets.all() == [pet1, pet2, pet3])
# remove child from parent
logger.debug('***********C - remove test_parent from pet3****************')
pet3.parents.remove(test_parent) ## ERROR here
logger.debug('**********C - done!***************')
# resultant elements should be remaining pets, and no child
self.assertTrue(test_parent.children.all() == [])
self.assertTrue(test_parent.pets.all() == [pet1, pet2]) # pet2 was not touched,
# but pet1 should remain
# since only
# pet3 was removed
# child1 should be also removed since
# relationship is unbalanced, i.e.
# user can't have access to a child if it
# does not have access to all of the child's pets
def test_child_pet_relationship_on_parents_combined_reversed(self):
"""
Test that a parent can hold children and pets that don't
belong necessary to the child.
"""
# create new parent
test_parent = Parent(name='test_parent')
child1 = Child(id=1,
name='FakeChild1')
child2 = Child(id=2,
name='FakeChild2')
pet1 = Pet(id=1,
name='FakePet1',
child_id=1)
pet2 = Pet(id=2,
name='FakePet2',
child_id=2)
pet3 = Pet(id=3,
name='FakePet3',
child_id=1)
self.session.add(test_parent)
self.session.add(child1)
self.session.add(child2)
self.session.add(pet1)
self.session.add(pet2)
self.session.add(pet3)
self.session.commit()
logger.debug('************A` - add child1 to test_parent***************')
# add parent to the child
test_parent.children.append(child1)
self.session.add(test_parent)
self.session.commit()
logger.debug('**********A` - done!***************')
logger.debug('************B` - add pet2 to test_parent***************')
# add parent to the child
test_parent.pets.append(pet2)
# persist changes in the db
self.session.add(test_parent)
self.session.commit()
logger.debug('**********B` - done!***************')
# check that previous relationships are intact
self.assertTrue(child1.pets.all() == [pet1, pet3])
self.assertTrue(child2.pets.all() == [pet2])
# resultant elements should be only child1, its pets and the single Pet
self.assertTrue(test_parent.children.all() == [child1])
self.assertTrue(test_parent.pets.all() == [pet1, pet2, pet3])
# remove child from parent
logger.debug('***********C` - remove pet3 from test_parent****************')
test_parent.pets.remove(pet3)
logger.debug('**********C` - done!***************')
# resultant elements should be remaining pets, and no child
self.assertTrue(test_parent.children.all() == [])
self.assertTrue(test_parent.pets.all() == [pet1, pet2]) # pet2 was not touched,
# but pet1 should remain
# since only
# pet3 was removed
# child1 should be also removed since
# relationship is unbalanced, i.e.
# user can't have access to a child if it
# does not have access to all of the child's pets
import sys
if __name__ == '__main__':
# # run tests
unittest.main()
第一次测试通过，但第二次没有。对于第一次测试，
2018-05-21 11:52:50,646:pets2.py:195 DEBUG ************A - add test_parent to child1***************
2018-05-21 11:52:50,648:base.py:682 INFO BEGIN (implicit)
2018-05-21 11:52:50,649:base.py:1151 INFO SELECT children.id AS children_id, children.name AS children_name
FROM children
WHERE children.id = ?
2018-05-21 11:52:50,649:base.py:1154 INFO (1,)
2018-05-21 11:52:50,650:result.py:681 DEBUG Col ('children_id', 'children_name')
2018-05-21 11:52:50,650:result.py:1106 DEBUG Row (1, 'FakeChild1')
2018-05-21 11:52:50,652:base.py:1151 INFO SELECT parents.id AS parents_id, parents.name AS parents_name
FROM parents
WHERE parents.id = ?
2018-05-21 11:52:50,652:base.py:1154 INFO (1,)
2018-05-21 11:52:50,652:result.py:681 DEBUG Col ('parents_id', 'parents_name')
2018-05-21 11:52:50,652:result.py:1106 DEBUG Row (1, 'test_parent')
2018-05-21 11:52:50,652:pets2.py:91 DEBUG **********1. adding the pets of <Child (name='FakeChild1')> to <Parent (name='test_parent')>***************
2018-05-21 11:52:50,654:base.py:1151 INFO INSERT INTO parents_children_relationship (parent_id, child_id) VALUES (?, ?)
2018-05-21 11:52:50,654:base.py:1154 INFO (1, 1)
2018-05-21 11:52:50,656:base.py:1151 INFO SELECT pets.id AS pets_id, pets.name AS pets_name, pets.child_id AS pets_child_id
FROM pets
WHERE ? = pets.child_id
2018-05-21 11:52:50,656:base.py:1154 INFO (1,)
2018-05-21 11:52:50,657:result.py:681 DEBUG Col ('pets_id', 'pets_name', 'pets_child_id')
2018-05-21 11:52:50,657:result.py:1106 DEBUG Row (1, 'FakePet1', 1)
2018-05-21 11:52:50,657:result.py:1106 DEBUG Row (3, 'FakePet3', 1)
2018-05-21 11:52:50,658:base.py:1151 INFO INSERT INTO parents_pets_relationship VALUES (?, ?)
2018-05-21 11:52:50,658:base.py:1154 INFO ((1, 1), (1, 3))
2018-05-21 11:52:50,658:pets2.py:102 DEBUG **********1. done!***************
2018-05-21 11:52:50,658:base.py:722 INFO COMMIT
2018-05-21 11:52:50,659:pets2.py:199 DEBUG **********A - done!***************
但是对于第二次测试，日志和错误回溯是冗长的，但失败的部分是
2018-05-21 11:52:50,723:pets2.py:274 DEBUG ************A` - add child1 to test_parent***************
2018-05-21 11:52:50,724:base.py:682 INFO BEGIN (implicit)
2018-05-21 11:52:50,724:base.py:1151 INFO SELECT children.id AS children_id, children.name AS children_name
FROM children
WHERE children.id = ?
2018-05-21 11:52:50,724:base.py:1154 INFO (1,)
2018-05-21 11:52:50,725:result.py:681 DEBUG Col ('children_id', 'children_name')
2018-05-21 11:52:50,725:result.py:1106 DEBUG Row (1, 'FakeChild1')
2018-05-21 11:52:50,726:base.py:1151 INFO SELECT parents.id AS parents_id, parents.name AS parents_name
FROM parents
WHERE parents.id = ?
2018-05-21 11:52:50,726:base.py:1154 INFO (1,)
2018-05-21 11:52:50,726:result.py:681 DEBUG Col ('parents_id', 'parents_name')
2018-05-21 11:52:50,727:result.py:1106 DEBUG Row (1, 'test_parent')
2018-05-21 11:52:50,727:pets2.py:91 DEBUG **********1. adding the pets of <Child (name='FakeChild1')> to <Parent (name='test_parent')>***************
2018-05-21 11:52:50,729:base.py:1151 INFO INSERT INTO parents_children_relationship (parent_id, child_id) VALUES (?, ?)
2018-05-21 11:52:50,729:base.py:1154 INFO (1, 1)
2018-05-21 11:52:50,731:base.py:1151 INFO SELECT pets.id AS pets_id, pets.name AS pets_name, pets.child_id AS pets_child_id
FROM pets
WHERE ? = pets.child_id
2018-05-21 11:52:50,731:base.py:1154 INFO (1,)
2018-05-21 11:52:50,731:result.py:681 DEBUG Col ('pets_id', 'pets_name', 'pets_child_id')
2018-05-21 11:52:50,732:result.py:1106 DEBUG Row (1, 'FakePet1', 1)
2018-05-21 11:52:50,732:result.py:1106 DEBUG Row (3, 'FakePet3', 1)
2018-05-21 11:52:50,732:base.py:1151 INFO INSERT INTO parents_pets_relationship VALUES (?, ?)
2018-05-21 11:52:50,732:base.py:1154 INFO ((1, 1), (1, 3))
2018-05-21 11:52:50,733:pets2.py:102 DEBUG **********1. done!***************
2018-05-21 11:52:50,735:base.py:1151 INFO INSERT INTO parents_children_relationship (parent_id, child_id) VALUES (?, ?)
2018-05-21 11:52:50,735:base.py:1154 INFO (1, 1)
2018-05-21 11:52:50,735:base.py:702 INFO ROLLBACK
基于记录器调用，在Parent.children.append的事件监听器完成之后且在调用pet2.parents_pets.append(test_parent)之前发生“某事”。这导致关联表中的一行两次，为什么会发生这种情况？
测试的特定错误回溯是
sqlalchemy.exc.IntegrityError: (sqlite3.IntegrityError) UNIQUE constraint failed: parents_children_relationship.parent_id, parents_children_relationship.child_id [SQL: 'INSERT INTO parents_children_relationship (parent_id, child_id) VALUES (?, ?)'] [parameters: (1, 1)]
这个问题与此有关
override relationship behaviour in sqlalchemy
还有Sqlalchemy邮件列表中的同一个问题
https://groups.google.com/forum/#!topic/sqlalchemy/jgKgv5zQT7E
我知道要处理很多，但我想知道为什么这种关系不能对称地工作，为什么一个测试通过而另一个不通过？
谢谢！
编辑：
如果我替换一个侦听对应的append方法的事件监听器（即，多对多关系中另一方的类的方法），对于像
@sa.event.listens_for(Child.parents_children, 'append')
def _on_append_children(child, parent, initiator):
"""
If a new child is appended to the parent, this listener
will also add the pets bound to the child being bound to the parent.
"""
# appends also the pets bound to the child that the
# parent is being appended to
logger.debug(f'**********1. (reversed) Adding the pets of {child} to {parent}***************')
object_session(child).execute(
"INSERT INTO parents_pets_relationship VALUES "
"(:parent_id, :pet_id)",
[
{"parent_id": parent.id, "pet_id": pet.id}
for pet in child.pets
]
)
logger.debug('**********1. done!***************')
然后第一次测试失败，第二次测试失败，这有点隔离了listeners的问题，可能与object_session对象有关。
来自SQLAlchemy的邮件列表（我为了上下文编辑了它，为@zzzeek提供了耐心和帮助！）
问题是因为向child1.parents添加test_parent会触发两个单独的“脏”事件，每个事件都会解析为parent_children_relationship表中的相同INSERT操作。通常，这两个脏事件在刷新过程中一起解析。但是，在“insert”事件处理程序中，调用“child.pets”集合，因为它是动态关系，会触发autoflush。因此，处理父项附加到子项的第一个脏事件，该行插入autoflush中。那么你的活动就完成了然后backref处理程序关闭并将child附加到parent，触发第二个脏事件。 session.commit（）然后尝试第二次刷新同样的事情失败。
解决方案是不在事件处理程序中刷新。
sess = object_session(parent) with sess.no_autoflush: sess.execute( "INSERT INTO parents_pets_relationship VALUES " "(:parent_id, :pet_id)", [ {"parent_id": parent.id, "pet_id": pet.id} for pet in child.pets ] )
在侦听器上更改此部分会使两个测试都通过