Раньше я думал, что во многих отношениях многие (возможно, для всех доступных отношений)
parent.children.append(child)
и
child.parents.append(parent)
были эквивалентны.У меня, однако, довольно сложная установка, которая запускает прослушиватели событий для append и remove методов моих классов.
ThisВозможно, это длинный путь, но я написал отдельный пример, который объясняет мою проблему.Короче говоря, он содержит:
- Три класса:
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 завершен и до вызова 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)]
Этот вопрос относится к
переопределить поведение отношений в sqlalchemy
А также к тому же вопросу в списке рассылки Sqlalchemy
https://groups.google.com/forum/#!topic/sqlalchemy/jgKgv5zQT7E
Я знаю, что многое нужно обработать, но мне интересно, почему эти отношения не работают симметрично, т.е. почему один тест пройден, а другой нет?
Спасибо!
РЕДАКТИРОВАТЬ:
Если я заменим прослушиватель событий на тот, который прослушивает метод counterpart 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.