diff --git a/django/db/models/base.py b/django/db/models/base.py index a65fd70302..6304e009d3 100644 --- a/django/db/models/base.py +++ b/django/db/models/base.py @@ -588,20 +588,22 @@ class Model(object): for related in self._meta.get_all_related_many_to_many_objects(): if related.field.rel.through: + db = router.db_for_write(related.field.rel.through.__class__, instance=self) opts = related.field.rel.through._meta reverse_field_name = related.field.m2m_reverse_field_name() nullable = opts.get_field(reverse_field_name).null filters = {reverse_field_name: self} - for sub_obj in related.field.rel.through._base_manager.filter(**filters): + for sub_obj in related.field.rel.through._base_manager.using(db).filter(**filters): sub_obj._collect_sub_objects(seen_objs, self, nullable) for f in self._meta.many_to_many: if f.rel.through: + db = router.db_for_write(f.rel.through.__class__, instance=self) opts = f.rel.through._meta field_name = f.m2m_field_name() nullable = opts.get_field(field_name).null filters = {field_name: self} - for sub_obj in f.rel.through._base_manager.filter(**filters): + for sub_obj in f.rel.through._base_manager.using(db).filter(**filters): sub_obj._collect_sub_objects(seen_objs, self, nullable) else: # m2m-ish but with no through table? GenericRelation: cascade delete @@ -627,7 +629,6 @@ class Model(object): def delete(self, using=None): using = using or router.db_for_write(self.__class__, instance=self) - connection = connections[using] assert self._get_pk_val() is not None, "%s object can't be deleted because its %s attribute is set to None." % (self._meta.object_name, self._meta.pk.attname) # Find all the objects than need to be deleted. diff --git a/tests/regressiontests/multiple_database/models.py b/tests/regressiontests/multiple_database/models.py index 9739fcffd7..e0be761ca0 100644 --- a/tests/regressiontests/multiple_database/models.py +++ b/tests/regressiontests/multiple_database/models.py @@ -44,6 +44,16 @@ class Book(models.Model): class Meta: ordering = ('title',) +class Pet(models.Model): + name = models.CharField(max_length=100) + owner = models.ForeignKey(Person) + + def __unicode__(self): + return self.name + + class Meta: + ordering = ('name',) + class UserProfile(models.Model): user = models.OneToOneField(User, null=True) flavor = models.CharField(max_length=100) diff --git a/tests/regressiontests/multiple_database/tests.py b/tests/regressiontests/multiple_database/tests.py index 98b203781d..7bde8bf037 100644 --- a/tests/regressiontests/multiple_database/tests.py +++ b/tests/regressiontests/multiple_database/tests.py @@ -10,7 +10,7 @@ from django.db import connections, router, DEFAULT_DB_ALIAS from django.db.utils import ConnectionRouter from django.test import TestCase -from models import Book, Person, Review, UserProfile +from models import Book, Person, Pet, Review, UserProfile try: # we only have these models if the user is using multi-db, it's safe the @@ -321,6 +321,66 @@ class QueryTestCase(TestCase): except ValueError: pass + def test_m2m_deletion(self): + "Cascaded deletions of m2m relations issue queries on the right database" + # Create a book and author on the other database + dive = Book.objects.using('other').create(title="Dive into Python", + published=datetime.date(2009, 5, 4)) + + mark = Person.objects.using('other').create(name="Mark Pilgrim") + dive.authors = [mark] + + # Check the initial state + self.assertEquals(Person.objects.using('default').count(), 0) + self.assertEquals(Book.objects.using('default').count(), 0) + self.assertEquals(Book.authors.through.objects.using('default').count(), 0) + + self.assertEquals(Person.objects.using('other').count(), 1) + self.assertEquals(Book.objects.using('other').count(), 1) + self.assertEquals(Book.authors.through.objects.using('other').count(), 1) + + # Delete the object on the other database + dive.delete(using='other') + + self.assertEquals(Person.objects.using('default').count(), 0) + self.assertEquals(Book.objects.using('default').count(), 0) + self.assertEquals(Book.authors.through.objects.using('default').count(), 0) + + # The person still exists ... + self.assertEquals(Person.objects.using('other').count(), 1) + # ... but the book has been deleted + self.assertEquals(Book.objects.using('other').count(), 0) + # ... and the relationship object has also been deleted. + self.assertEquals(Book.authors.through.objects.using('other').count(), 0) + + # Now try deletion in the reverse direction. Set up the relation again + dive = Book.objects.using('other').create(title="Dive into Python", + published=datetime.date(2009, 5, 4)) + dive.authors = [mark] + + # Check the initial state + self.assertEquals(Person.objects.using('default').count(), 0) + self.assertEquals(Book.objects.using('default').count(), 0) + self.assertEquals(Book.authors.through.objects.using('default').count(), 0) + + self.assertEquals(Person.objects.using('other').count(), 1) + self.assertEquals(Book.objects.using('other').count(), 1) + self.assertEquals(Book.authors.through.objects.using('other').count(), 1) + + # Delete the object on the other database + mark.delete(using='other') + + self.assertEquals(Person.objects.using('default').count(), 0) + self.assertEquals(Book.objects.using('default').count(), 0) + self.assertEquals(Book.authors.through.objects.using('default').count(), 0) + + # The person has been deleted ... + self.assertEquals(Person.objects.using('other').count(), 0) + # ... but the book still exists + self.assertEquals(Book.objects.using('other').count(), 1) + # ... and the relationship object has been deleted. + self.assertEquals(Book.authors.through.objects.using('other').count(), 0) + def test_foreign_key_separation(self): "FK fields are constrained to a single database" # Create a book and author on the default database @@ -498,6 +558,28 @@ class QueryTestCase(TestCase): self.assertEquals(list(Book.objects.using('other').values_list('title',flat=True)), [u'Dive into HTML5', u'Dive into Python', u'Dive into Water']) + def test_foreign_key_deletion(self): + "Cascaded deletions of Foreign Key relations issue queries on the right database" + mark = Person.objects.using('other').create(name="Mark Pilgrim") + fido = Pet.objects.using('other').create(name="Fido", owner=mark) + + # Check the initial state + self.assertEquals(Person.objects.using('default').count(), 0) + self.assertEquals(Pet.objects.using('default').count(), 0) + + self.assertEquals(Person.objects.using('other').count(), 1) + self.assertEquals(Pet.objects.using('other').count(), 1) + + # Delete the person object, which will cascade onto the pet + mark.delete(using='other') + + self.assertEquals(Person.objects.using('default').count(), 0) + self.assertEquals(Pet.objects.using('default').count(), 0) + + # Both the pet and the person have been deleted from the right database + self.assertEquals(Person.objects.using('other').count(), 0) + self.assertEquals(Pet.objects.using('other').count(), 0) + def test_o2o_separation(self): "OneToOne fields are constrained to a single database" # Create a user and profile on the default database @@ -729,6 +811,29 @@ class QueryTestCase(TestCase): self.assertEquals(list(Review.objects.using('other').filter(object_id=dive.pk).values_list('source',flat=True)), [u'Python Daily', u'Python Weekly']) + def test_generic_key_deletion(self): + "Cascaded deletions of Generic Key relations issue queries on the right database" + dive = Book.objects.using('other').create(title="Dive into Python", + published=datetime.date(2009, 5, 4)) + review = Review.objects.using('other').create(source="Python Weekly", content_object=dive) + + # Check the initial state + self.assertEquals(Book.objects.using('default').count(), 0) + self.assertEquals(Review.objects.using('default').count(), 0) + + self.assertEquals(Book.objects.using('other').count(), 1) + self.assertEquals(Review.objects.using('other').count(), 1) + + # Delete the Book object, which will cascade onto the pet + dive.delete(using='other') + + self.assertEquals(Book.objects.using('default').count(), 0) + self.assertEquals(Review.objects.using('default').count(), 0) + + # Both the pet and the person have been deleted from the right database + self.assertEquals(Book.objects.using('other').count(), 0) + self.assertEquals(Review.objects.using('other').count(), 0) + def test_ordering(self): "get_next_by_XXX commands stick to a single database" pro = Book.objects.create(title="Pro Django",