Loading django/db/models/query.py +1 −0 Original line number Diff line number Diff line Loading @@ -742,6 +742,7 @@ class QuerySet(object): # Default to false for nowait nowait = kwargs.pop('nowait', False) obj = self._clone() obj._for_write = True obj.query.select_for_update = True obj.query.select_for_update_nowait = nowait return obj Loading tests/select_for_update/tests.py +13 −1 Original line number Diff line number Diff line Loading @@ -5,11 +5,13 @@ import time import unittest from django.conf import settings from django.db import transaction, connection from django.db import transaction, connection, router from django.db.utils import ConnectionHandler, DEFAULT_DB_ALIAS, DatabaseError from django.test import (TransactionTestCase, skipIfDBFeature, skipUnlessDBFeature) from multiple_database.tests import TestRouter from .models import Person # Some tests require threading, which might not be available. So create a Loading Loading @@ -254,3 +256,13 @@ class SelectForUpdateTests(TransactionTestCase): """ people = list(Person.objects.select_for_update()) self.assertTrue(transaction.is_dirty()) @skipUnlessDBFeature('has_select_for_update') def test_select_for_update_on_multidb(self): old_routers = router.routers try: router.routers = [TestRouter()] query = Person.objects.select_for_update() self.assertEqual(router.db_for_write(Person), query.db) finally: router.routers = old_routers Loading
django/db/models/query.py +1 −0 Original line number Diff line number Diff line Loading @@ -742,6 +742,7 @@ class QuerySet(object): # Default to false for nowait nowait = kwargs.pop('nowait', False) obj = self._clone() obj._for_write = True obj.query.select_for_update = True obj.query.select_for_update_nowait = nowait return obj Loading
tests/select_for_update/tests.py +13 −1 Original line number Diff line number Diff line Loading @@ -5,11 +5,13 @@ import time import unittest from django.conf import settings from django.db import transaction, connection from django.db import transaction, connection, router from django.db.utils import ConnectionHandler, DEFAULT_DB_ALIAS, DatabaseError from django.test import (TransactionTestCase, skipIfDBFeature, skipUnlessDBFeature) from multiple_database.tests import TestRouter from .models import Person # Some tests require threading, which might not be available. So create a Loading Loading @@ -254,3 +256,13 @@ class SelectForUpdateTests(TransactionTestCase): """ people = list(Person.objects.select_for_update()) self.assertTrue(transaction.is_dirty()) @skipUnlessDBFeature('has_select_for_update') def test_select_for_update_on_multidb(self): old_routers = router.routers try: router.routers = [TestRouter()] query = Person.objects.select_for_update() self.assertEqual(router.db_for_write(Person), query.db) finally: router.routers = old_routers