我曾经认为，在多对多关系中(也许对于所有可用的关系(
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)dt%(levelname)st%(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)]
这个问题与
覆盖 sqllchemy 中的关系行为
以及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的耐心和帮助！
问题是因为给孩子增加了test_parent 1.父母 触发两个单独的"脏"事件，每个事件解析为相同的 将操作插入到parents_children_relationship表中。 通常，这两个脏事件在 冲洗过程。但是，在"插入"事件处理程序中，调用 在"Child.Pets"系列上，由于它是一种动态关系， 触发自动刷新。 因此，父母的第一个肮脏事件是 处理追加到子项，在自动刷新中插入该行。 然后您的活动完成。 然后 backref 处理程序关闭并 将子项追加到父项，触发第二个脏事件。 这 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 ] )
在侦听器上更改此部分会使两个测试都通过