Author: russellm
Date: 2010-05-11 08:06:03 -0500 (Tue, 11 May 2010)
New Revision: 13232
Modified:
django/trunk/django/db/models/base.py
django/trunk/tests/regressiontests/multiple_database/models.py
django/trunk/tests/regressiontests/multiple_database/tests.py
Log:
Fixed #13513 -- Ensured that queries collecting deleted objects are issued on
the right database, especially when dealing with m2m intermediate tables.
Thanks to gavoja for the report.
Modified: django/trunk/django/db/models/base.py
===================================================================
--- django/trunk/django/db/models/base.py 2010-05-11 06:10:58 UTC (rev
13231)
+++ django/trunk/django/db/models/base.py 2010-05-11 13:06:03 UTC (rev
13232)
@@ -588,20 +588,22 @@
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 @@
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.
Modified: django/trunk/tests/regressiontests/multiple_database/models.py
===================================================================
--- django/trunk/tests/regressiontests/multiple_database/models.py
2010-05-11 06:10:58 UTC (rev 13231)
+++ django/trunk/tests/regressiontests/multiple_database/models.py
2010-05-11 13:06:03 UTC (rev 13232)
@@ -44,6 +44,16 @@
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)
Modified: django/trunk/tests/regressiontests/multiple_database/tests.py
===================================================================
--- django/trunk/tests/regressiontests/multiple_database/tests.py
2010-05-11 06:10:58 UTC (rev 13231)
+++ django/trunk/tests/regressiontests/multiple_database/tests.py
2010-05-11 13:06:03 UTC (rev 13232)
@@ -10,7 +10,7 @@
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 @@
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 @@
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 @@
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",
--
You received this message because you are subscribed to the Google Groups
"Django updates" group.
To post to this group, send email to [email protected].
To unsubscribe from this group, send email to
[email protected].
For more options, visit this group at
http://groups.google.com/group/django-updates?hl=en.
