backref 关系中的对称性无法按预期与事件侦听器一起工作
symmetry in backref relationships not working as expected with event listeners
我曾经认为，在多对多关系中（也许对于所有可用的关系）
parent.children.append(child)
和
child.parents.append(parent)
是等价的。但是，我有一个相当 复杂的 设置，它会为 append 和 remove 触发事件侦听器我的方法 类.
这可能是一个远景，但我已经写了一个独立的例子来解释我的问题。简而言之，它包含：
- 三个 类：
Parent、Child 和 Pet。在哪里
Parent 与 Child 和 PetChild与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 是Parenttest_child_pet_relationship_on_parents_combined_reversed 做同样的事情，但反过来，像 test_parent.children.append(child1).
- A
logging 设置记录每个查询发生的任何事情。我在每个测试以及第一个事件侦听器中都设置了一些草率的记录器条目。即，带有 A 的日志指的是第一个测试，A' 指的是第二个测试。
asas
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 的事件侦听器完成并且before pet2.parents_pets.append(test_parent) 被调用。这导致在关联中连续 table 两次，为什么会这样？
测试的具体错误回溯是
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)]
这个问题与
相关
还有 Sqlalchemy 邮件列表中的相同问题
https://groups.google.com/forum/#!topic/sqlalchemy/jgKgv5zQT7E
我知道有很多事情要处理，但我想知道为什么这种关系不能对称地工作，即为什么一个测试通过而另一个没有通过？
谢谢！
编辑：
如果我将事件侦听器替换为侦听 counterpartappend 方法的事件侦听器（即 Class 的方法，即多对多关系中的另一方），例如
@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 的耐心和帮助！）
the issue is because the addition of test_parent to child1.parents
triggers two individual "dirty" events, which each resolve to the same
INSERT operation into the parents_children_relationship table.
Normally, these two dirty events are resolved together during the
flush process. However, within your "insert" event handler, calling
upon the "child.pets" collection, since it is a dynamic relationship,
triggers autoflush. So, the first dirty event for parent being
appended to child is handled, the row is inserted within autoflush.
then your event completes. then the backref handler goes off and
appends child to parent, triggering the second dirty event. the
session.commit() then tries to flush the same thing a second time
which fails.
the solution is to not flush within the event handlers.
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
]
)
在侦听器上更改此部分会使两个测试都通过
我曾经认为，在多对多关系中（也许对于所有可用的关系）
parent.children.append(child)
和
child.parents.append(parent)
是等价的。但是，我有一个相当 复杂的 设置，它会为 append 和 remove 触发事件侦听器我的方法 类.
这可能是一个远景，但我已经写了一个独立的例子来解释我的问题。简而言之，它包含：
- 三个 类：
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).
- A
logging设置记录每个查询发生的任何事情。我在每个测试以及第一个事件侦听器中都设置了一些草率的记录器条目。即，带有A的日志指的是第一个测试，A'指的是第二个测试。
asas
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 的事件侦听器完成并且before pet2.parents_pets.append(test_parent) 被调用。这导致在关联中连续 table 两次，为什么会这样？
测试的具体错误回溯是
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)]
这个问题与
相关还有 Sqlalchemy 邮件列表中的相同问题
https://groups.google.com/forum/#!topic/sqlalchemy/jgKgv5zQT7E
我知道有很多事情要处理，但我想知道为什么这种关系不能对称地工作，即为什么一个测试通过而另一个没有通过？
谢谢！
编辑：
如果我将事件侦听器替换为侦听 counterpartappend 方法的事件侦听器（即 Class 的方法，即多对多关系中的另一方），例如
@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 的耐心和帮助！）
the issue is because the addition of test_parent to child1.parents triggers two individual "dirty" events, which each resolve to the same INSERT operation into the parents_children_relationship table. Normally, these two dirty events are resolved together during the flush process. However, within your "insert" event handler, calling upon the "child.pets" collection, since it is a dynamic relationship, triggers autoflush. So, the first dirty event for parent being appended to child is handled, the row is inserted within autoflush. then your event completes. then the backref handler goes off and appends child to parent, triggering the second dirty event. the session.commit() then tries to flush the same thing a second time which fails.
the solution is to not flush within the event handlers.
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 ] )
在侦听器上更改此部分会使两个测试都通过