mirrors/django
1
0
Fork 0
mirror of https://github.com/django/django.git synced 2024-12-22 09:05:43 +00:00
Code Issues 25 Releases Wiki Activity

Fixed #27062 -- Eased implementing select_for_update() on MSSQL.

Browse Source
This commit is contained in:
Mikhail Denisenko 2015-11-15 23:14:55 -05:00 committed by Tim Graham
parent ef021412d5
commit bae64dd0f1
3 changed files with 36 additions and 18 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

3
django/db/backends/base/features.py
View File

@ -229,6 +229,9 @@ class BaseDatabaseFeatures(object):
# be equal?
ignores_quoted_identifier_case = False
# Place FOR UPDATE right after FROM clause. Used on MSSQL.
for_update_after_from = False
def __init__(self, connection):
self.connection = connection

37
django/db/models/sql/compiler.py
View File

@ -402,6 +402,25 @@ class SQLCompiler(object):
result.extend(from_)
params.extend(f_params)
for_update_part = None
if self.query.select_for_update and self.connection.features.has_select_for_update:
if self.connection.get_autocommit():
raise TransactionManagementError("select_for_update cannot be used outside of a transaction.")
nowait = self.query.select_for_update_nowait
skip_locked = self.query.select_for_update_skip_locked
# If it's a NOWAIT/SKIP LOCKED query but the backend doesn't
# support it, raise a DatabaseError to prevent a possible
# deadlock.
if nowait and not self.connection.features.has_select_for_update_nowait:
raise DatabaseError('NOWAIT is not supported on this database backend.')
elif skip_locked and not self.connection.features.has_select_for_update_skip_locked:
raise DatabaseError('SKIP LOCKED is not supported on this database backend.')
for_update_part = self.connection.ops.for_update_sql(nowait=nowait, skip_locked=skip_locked)
if for_update_part and self.connection.features.for_update_after_from:
result.append(for_update_part)
if where:
result.append('WHERE %s' % where)
params.extend(w_params)
@ -439,22 +458,8 @@ class SQLCompiler(object):
result.append('LIMIT %d' % val)
result.append('OFFSET %d' % self.query.low_mark)
if self.query.select_for_update and self.connection.features.has_select_for_update:
if self.connection.get_autocommit():
raise TransactionManagementError(
"select_for_update cannot be used outside of a transaction."
)
nowait = self.query.select_for_update_nowait
skip_locked = self.query.select_for_update_skip_locked
# If we've been asked for a NOWAIT/SKIP LOCKED query but the
# backend does not support it, raise a DatabaseError otherwise
# we could get an unexpected deadlock.
if nowait and not self.connection.features.has_select_for_update_nowait:
raise DatabaseError('NOWAIT is not supported on this database backend.')
elif skip_locked and not self.connection.features.has_select_for_update_skip_locked:
raise DatabaseError('SKIP LOCKED is not supported on this database backend.')
result.append(self.connection.ops.for_update_sql(nowait=nowait, skip_locked=skip_locked))
if for_update_part and not self.connection.features.for_update_after_from:
result.append(for_update_part)
return ' '.join(result), tuple(params)
finally:

14
tests/select_for_update/tests.py
View File

@ -5,9 +5,11 @@ import time
from multiple_database.routers import TestRouter
from django.db import DatabaseError, connection, router, transaction
from django.db import (
DatabaseError, connection, connections, router, transaction,
)
from django.test import (
TransactionTestCase, override_settings, skipIfDBFeature,
TransactionTestCase, mock, override_settings, skipIfDBFeature,
skipUnlessDBFeature,
)
from django.test.utils import CaptureQueriesContext
@ -150,6 +152,14 @@ class SelectForUpdateTests(TransactionTestCase):
with transaction.atomic():
Person.objects.select_for_update(skip_locked=True).get()
@skipUnlessDBFeature('has_select_for_update')
def test_for_update_after_from(self):
features_class = connections['default'].features.__class__
attribute_to_patch = "%s.%s.for_update_after_from" % (features_class.__module__, features_class.__name__)
with mock.patch(attribute_to_patch, return_value=True):
with transaction.atomic():
self.assertIn('FOR UPDATE WHERE', str(Person.objects.filter(name='foo').select_for_update().query))
@skipUnlessDBFeature('has_select_for_update')
def test_for_update_requires_transaction(self):
"""