Django raw SQL and transactions

last modified November 13, 2026 Python 3.15

In this article we leave the ORM when it does not fit and keep the data consistent. We run raw SQL with raw() and with connection.cursor(), pass parameters safely, group changes into transactions with transaction.atomic(), nest them with savepoints, run code after a commit with on_commit(), and wrap every request in a transaction with ATOMIC_REQUESTS. The examples use Django 6.1, Python 3.15, and the default SQLite database.

When to use raw SQL

The ORM covers most queries, and it protects us from SQL injection and from differences between databases. Sometimes it is not enough: a complex report with window functions, a vendor-specific feature, or a query that the ORM translates badly. Django then lets us write SQL by hand. There are two levels. The raw() method of a manager runs a SELECT statement and returns model instances. The connection.cursor() function gives us a low-level database cursor that can run any statement and returns plain tuples. Before writing raw SQL, check whether aggregation and expressions can do the job; they keep the query portable.

The example project

The project is created with uv; the steps are explained in Django introduction. The commands below create a project directory, install Django, generate the project with the configuration package config, start the app bank, and create the folder demos for demo scripts.

$ mkdir transactions
$ cd transactions
$ uv init --bare
$ uv add django
$ uv run django-admin startproject config .
$ uv run python manage.py startapp bank
$ mkdir demos

The trailing dot in startproject config . creates manage.py and the config package in the current directory instead of one level deeper. A new app is ignored by Django until it is listed in INSTALLED_APPS, so we add the app to the list in config/settings.py; the other entries are generated by startproject.

INSTALLED_APPS = [
    'django.contrib.admin',
    'django.contrib.auth',
    'django.contrib.contenttypes',
    'django.contrib.sessions',
    'django.contrib.messages',
    'django.contrib.staticfiles',
    'bank',
]

The listings in this article are of two kinds. A listing with a solid blue border and the label Project file is a file of the application: models, views, templates, tests, and settings. A listing with a dashed amber border and the label Demo script is a small helper in the demos folder; it sets up Django, runs a few lines of code, and prints the results, so that we can watch the behaviour. Demo scripts are not part of the application; we run one from the project directory with uv run python -m demos.name.

An account has an owner and a balance, and a CheckConstraint forbids a negative balance, so the database itself guards the rule. A transfer records the movement of money between two accounts.

from django.db import models
from django.db.models import Q


class Account(models.Model):
    owner = models.CharField(max_length=50)
    balance = models.IntegerField(default=0)

    class Meta:
        constraints = [
            models.CheckConstraint(condition=Q(balance__gte=0), name='balance_not_negative'),
        ]

    def __str__(self):
        return f'{self.owner}: {self.balance}'


class Transfer(models.Model):
    source = models.ForeignKey(Account, on_delete=models.CASCADE, related_name='outgoing')
    target = models.ForeignKey(Account, on_delete=models.CASCADE, related_name='incoming')
    amount = models.PositiveIntegerField()

The constraint is created by the migration (see Django migrations); it is the reason why one of the statements below fails later. A script creates three accounts.

import os

import django

os.environ.setdefault('DJANGO_SETTINGS_MODULE', 'config.settings')
django.setup()

from bank.models import Account

for owner, balance in [('alice', 500), ('bob', 200), ('carol', 50)]:
    Account.objects.create(owner=owner, balance=balance)

print(Account.objects.count(), 'accounts')

The balances are whole numbers to keep the arithmetic simple. After the migration, the accounts of alice, bob, and carol hold 500, 200, and 50.

$ uv run python manage.py makemigrations bank
Migrations for 'bank':
  bank/migrations/0001_initial.py
    + Create model Account
    + Create model Transfer
$ uv run python manage.py migrate bank
Operations to perform:
  Apply all migrations: bank
Running migrations:
  Applying bank.0001_initial... OK
$ uv run python -m demos.seed
3 accounts

The raw() method

The raw() method takes a SQL string and a list of parameters, and returns a RawQuerySet of model instances. Placeholders are written as %s, whatever the database driver is. The query must select the primary key. The columns are matched to the fields by name; a field that is not selected is deferred and loaded by an extra query when it is read. Extra columns become attributes of the instances, and the translations argument maps column names to field names.

import os

import django

os.environ.setdefault('DJANGO_SETTINGS_MODULE', 'config.settings')
django.setup()

from bank.models import Account

rich = Account.objects.raw('SELECT * FROM bank_account WHERE balance > %s ORDER BY balance', [100])
print([str(account) for account in rich])

names = Account.objects.raw('SELECT id, owner FROM bank_account ORDER BY id')
for account in names:
    print(account.owner, account.balance)

renamed = Account.objects.raw(
    'SELECT id, owner AS full_name, balance * 2 AS doubled FROM bank_account ORDER BY id',
    translations={'full_name': 'owner'})
print([(account.owner, account.doubled) for account in renamed])

attack = "alice' OR '1'='1"
print(len(list(Account.objects.raw('SELECT * FROM bank_account WHERE owner = %s', [attack]))))
print(len(list(Account.objects.raw(f"SELECT * FROM bank_account WHERE owner = '{attack}'"))))

The first query uses a parameter and returns two accounts. The second one selects only id and owner; reading balance works but runs a query for every account. The third query computes a doubled column and renames full_name to the field owner with translations. The last two lines test an injection attempt. The value is passed as a parameter, so the database treats it as one string, and it matches no rows. When the same value is pasted into the SQL text with an f-string, the OR '1'='1' becomes a part of the statement and returns all three accounts. Never build SQL from user input by formatting; always use parameters.

$ uv run python -m demos.raw
['bob: 200', 'alice: 500']
alice 500
bob 200
carol 50
[('alice', 1000), ('bob', 400), ('carol', 100)]
0
3

The database cursor

For statements that are not a model SELECT, we use a cursor from django.db.connection. It is a context manager that is closed at the end of the block. The execute() method runs a statement with parameters, fetchone() and fetchall() return tuples, and executemany() runs a statement for a list of parameter tuples. The description attribute holds the column names, which lets us build dictionaries. Table names cannot be parameters; the script takes the name from Account._meta.db_table.

import os

import django

os.environ.setdefault('DJANGO_SETTINGS_MODULE', 'config.settings')
django.setup()

from django.db import connection

from bank.models import Account

table = Account._meta.db_table

with connection.cursor() as cursor:
    cursor.execute(f'SELECT owner, balance FROM {table} WHERE balance >= %s ORDER BY balance DESC', [100])
    print(cursor.fetchall())

    cursor.execute(f'SELECT COUNT(*), SUM(balance) FROM {table}')
    print(cursor.fetchone())

    cursor.execute(f'SELECT id, owner FROM {table} ORDER BY id')
    columns = [column[0] for column in cursor.description]
    print([dict(zip(columns, row)) for row in cursor.fetchall()])

    cursor.execute('CREATE TEMP TABLE audit (owner TEXT, note TEXT)')
    cursor.executemany('INSERT INTO audit VALUES (%s, %s)',
                       [('alice', 'opened'), ('bob', 'opened'), ('carol', 'opened')])
    print('rows inserted:', cursor.rowcount)
    cursor.execute('SELECT * FROM audit WHERE owner <> %s', ['bob'])
    print(cursor.fetchall())

The script selects accounts with at least 100, aggregates the balances, builds dictionaries from the rows, and finally creates a temporary table and fills it with executemany(). The rowcount attribute holds the number of rows affected by the last statement, 3 here. Other databases may return -1 for statements that do not report it.

$ uv run python -m demos.cursor
[('alice', 500), ('bob', 200)]
(3, 750)
[{'id': 1, 'owner': 'alice'}, {'id': 2, 'owner': 'bob'}, {'id': 3, 'owner': 'carol'}]
rows inserted: 3
[('alice', 'opened'), ('carol', 'opened')]

Transactions and atomic()

A transaction groups statements so that they either all take effect or none does. Django runs in autocommit mode: every statement is committed at once, unless we open a transaction. The transaction.atomic() function marks a block, and it can be used as a context manager or as a decorator. When the block ends normally, the changes are committed; when an exception leaves the block, they are rolled back. The module below has two versions of a money transfer. The unsafe one debits the source account, then looks up the target, which may not exist. The safe one is decorated with @transaction.atomic and checks the balance first.

from django.db import transaction

from .models import Account, Transfer


class InsufficientFunds(Exception):
    pass


def transfer_unsafe(source_id, target_id, amount):
    source = Account.objects.get(pk=source_id)
    source.balance -= amount
    source.save()
    target = Account.objects.get(pk=target_id)
    target.balance += amount
    target.save()
    Transfer.objects.create(source=source, target=target, amount=amount)


@transaction.atomic
def transfer(source_id, target_id, amount):
    source = Account.objects.select_for_update().get(pk=source_id)
    target = Account.objects.select_for_update().get(pk=target_id)
    if source.balance < amount:
        raise InsufficientFunds(f'{source.owner} has only {source.balance}')
    source.balance -= amount
    source.save(update_fields=['balance'])
    target.balance += amount
    target.save(update_fields=['balance'])
    Transfer.objects.create(source=source, target=target, amount=amount)

If transfer_unsafe() fails on the second lookup, the money has already been taken from the source and is lost. The transfer() function is atomic: any exception, including Account.DoesNotExist or our InsufficientFunds, undoes every change made inside. The select_for_update() calls lock the rows until the end of the transaction on databases that support it.

import os

import django

os.environ.setdefault('DJANGO_SETTINGS_MODULE', 'config.settings')
django.setup()

from bank.models import Account
from bank.services import InsufficientFunds, transfer, transfer_unsafe


def show(label):
    print(f'{label:34}', ', '.join(str(a) for a in Account.objects.order_by('pk')))


show('start')

try:
    transfer_unsafe(1, 999, 100)
except Account.DoesNotExist:
    print('transfer_unsafe to a missing account failed')
show('after the failed unsafe transfer')

try:
    transfer(1, 999, 100)
except Account.DoesNotExist:
    print('transfer to a missing account failed')
show('after the failed atomic transfer')

try:
    transfer(3, 1, 100)
except InsufficientFunds as error:
    print('InsufficientFunds:', error)

transfer(1, 2, 120)
show('after transfer(1, 2, 120)')

The script prints the balances after each step. The unsafe transfer to the account 999 fails, but alice has lost 100: the balance dropped from 500 to 400. The atomic version fails in the same way and leaves everything as it was. The transfer from carol raises InsufficientFunds, and the last transfer of 120 from alice to bob succeeds. Catch exceptions outside of the atomic block; catching them inside would hide the error from Django, and the transaction would be committed.

$ uv run python -m demos.atomic
start                              alice: 500, bob: 200, carol: 50
transfer_unsafe to a missing account failed
after the failed unsafe transfer   alice: 400, bob: 200, carol: 50
transfer to a missing account failed
after the failed atomic transfer   alice: 400, bob: 200, carol: 50
InsufficientFunds: carol has only 50
after transfer(1, 2, 120)          alice: 280, bob: 320, carol: 50

Savepoints

Atomic blocks can be nested. The outermost block starts the real transaction, and each inner block creates a savepoint. When an exception leaves an inner block, the database rolls back to its savepoint, and the outer block can carry on and commit the rest. This is useful when a part of the work is allowed to fail, such as an insert that may violate a unique constraint.

import os

import django

os.environ.setdefault('DJANGO_SETTINGS_MODULE', 'config.settings')
django.setup()

from django.db import IntegrityError, transaction
from django.db.models import F

from bank.models import Account

with transaction.atomic():
    Account.objects.filter(owner='alice').update(balance=F('balance') + 10)
    try:
        with transaction.atomic():
            Account.objects.filter(owner='bob').update(balance=-5)
    except IntegrityError as error:
        print('inner block rolled back:', error)
    Account.objects.filter(owner='carol').update(balance=F('balance') + 10)

print(*Account.objects.order_by('pk'), sep=', ')

The outer block adds 10 to alice. The inner block sets the balance of bob to -5, which breaks the check constraint and raises IntegrityError; only the inner changes are rolled back. We catch the error outside of the inner block. After that, carol gets 10 too, and the outer block is committed: alice and carol have the extra 10, whereas bob is unchanged. The error message comes from the database.

$ uv run python -m demos.savepoints
inner block rolled back: CHECK constraint failed: balance_not_negative
alice: 290, bob: 320, carol: 60

select_for_update()

Two processes that read a balance and then write it back can overwrite each other's changes. The select_for_update() method adds FOR UPDATE to the query, so that the selected rows are locked until the end of the transaction, and other transactions that want to lock them wait. It works only inside an atomic block. It is supported by PostgreSQL, MySQL, and MariaDB; SQLite does not have row locks, and Django ignores the method there. Another way to avoid the race is to let the database do the arithmetic with an F expression in one UPDATE.

import os

import django

os.environ.setdefault('DJANGO_SETTINGS_MODULE', 'config.settings')
django.setup()

from django.db import connection

from bank.models import Account

print(Account.objects.select_for_update().filter(pk=1).query)
print('has_select_for_update:', connection.features.has_select_for_update)

The script prints the SQL, which has no FOR UPDATE on SQLite, and the has_select_for_update feature flag of the connection, which is False. On PostgreSQL the same query ends with FOR UPDATE.

$ uv run python -m demos.lock
SELECT "bank_account"."id", "bank_account"."owner", "bank_account"."balance" FROM "bank_account" WHERE "bank_account"."id" = 1
has_select_for_update: False

on_commit()

Some work must happen only after the data is safely stored: sending an email, enqueuing a background job, or invalidating a cache. If we do it inside the transaction, the action may be performed for data that is later rolled back, or a worker may look for a row that is not committed yet. The transaction.on_commit() function registers a callback that runs after the current transaction commits successfully. When the transaction is rolled back, the callback is dropped. Outside a transaction the callback runs at once. The sending of messages is covered in Django email.

import os

import django

os.environ.setdefault('DJANGO_SETTINGS_MODULE', 'config.settings')
django.setup()

from django.db import transaction

from bank.models import Account


def notify(message):
    print('on_commit:', message)


with transaction.atomic():
    Account.objects.create(owner='dave', balance=10)
    transaction.on_commit(lambda: notify('dave was created'))
    print('inside the atomic block')
print('after the atomic block')

try:
    with transaction.atomic():
        Account.objects.create(owner='erin', balance=10)
        transaction.on_commit(lambda: notify('erin was created'))
        raise RuntimeError('rollback')
except RuntimeError:
    print('rolled back, erin exists:', Account.objects.filter(owner='erin').exists())

transaction.on_commit(lambda: notify('outside of a transaction runs at once'))

The first block creates an account and registers a callback; the output shows that the callback runs after the block, not at the line where it was registered. In the second block an exception rolls back the creation of erin, the callback never runs, and the account does not exist. The last callback is registered outside of any transaction and runs immediately.

$ uv run python -m demos.on_commit
inside the atomic block
on_commit: dave was created
after the atomic block
rolled back, erin exists: False
on_commit: outside of a transaction runs at once

ATOMIC_REQUESTS

Instead of decorating every function, Django can wrap each view call in a transaction. We turn this on with the ATOMIC_REQUESTS database setting. A response that comes back normally commits the transaction; an exception from the view rolls it back. The price is that every request holds a transaction, also the ones that only read, so the setting suits applications where most views write. A view can opt out with the @transaction.non_atomic_requests decorator.

DATABASES = {
    'default': {
        'ENGINE': 'django.db.backends.sqlite3',
        'NAME': BASE_DIR / 'db.sqlite3',
        'ATOMIC_REQUESTS': True,
    }
}

The setting is a key of the database dictionary in settings.py. The project below has two views that both add 1000 to an account and then fail.

from django.db import transaction
from django.db.models import F
from django.http import HttpResponse

from .models import Account


def deposit_then_fail(request):
    Account.objects.filter(owner='alice').update(balance=F('balance') + 1000)
    raise RuntimeError('something went wrong after the update')


@transaction.non_atomic_requests
def deposit_then_fail_no_transaction(request):
    Account.objects.filter(owner='bob').update(balance=F('balance') + 1000)
    raise RuntimeError('something went wrong after the update')


def ok(request):
    return HttpResponse('ok')

The first view runs inside the request transaction, because of the setting; the second one opts out. Both raise RuntimeError, so the client gets a 500 response. The URLconf of the app maps the two paths to the views.

from django.urls import path

from . import views

urlpatterns = [
    path('', views.ok),
    path('atomic/', views.deposit_then_fail),
    path('non-atomic/', views.deposit_then_fail_no_transaction),
]

The project's URLconf includes it. We add the include() import and one route to the generated file.

"""
URL configuration for config project.

The `urlpatterns` list routes URLs to views. For more information please see:
    https://docs.djangoproject.com/en/6.1/topics/http/urls/
Examples:
Function views
    1. Add an import:  from my_app import views
    2. Add a URL to urlpatterns:  path('', views.home, name='home')
Class-based views
    1. Add an import:  from other_app.views import Home
    2. Add a URL to urlpatterns:  path('', Home.as_view(), name='home')
Including another URLconf
    1. Import the include() function: from django.urls import include, path
    2. Add a URL to urlpatterns:  path('blog/', include('blog.urls'))
"""
from django.contrib import admin
from django.urls import include, path

urlpatterns = [
    path('admin/', admin.site.urls),
    path('', include('bank.urls')),
]

To see the effect we need a way to print the balances. The demo script lists the accounts in the order of their keys.

import os

import django

os.environ.setdefault('DJANGO_SETTINGS_MODULE', 'config.settings')
django.setup()

from bank.models import Account

print(*Account.objects.order_by('pk'), sep=', ')

The following commands print the accounts, call both URLs of the running development server, and print the accounts again. The 1000 added to alice was rolled back, but the 1000 added to bob stayed.

$ uv run python -m demos.show
alice: 290, bob: 320, carol: 60, dave: 10
$ curl -s -o /dev/null -w "%{http_code}\n" http://127.0.0.1:8000/atomic/
500
$ curl -s -o /dev/null -w "%{http_code}\n" http://127.0.0.1:8000/non-atomic/
500
$ uv run python -m demos.show
alice: 290, bob: 1320, carol: 60, dave: 10

Testing transactions

The test case checks the transfer functions, the check constraint, and the on_commit() callback. Django wraps every TestCase test in a transaction that is rolled back at the end, so a real commit never happens and the registered callbacks would not run; the captureOnCommitCallbacks() context manager collects them, and with execute=True runs them when the block ends.

from django.db import IntegrityError, transaction
from django.test import TestCase

from .models import Account, Transfer
from .services import InsufficientFunds, transfer


class TransferTests(TestCase):

    def setUp(self):
        self.alice = Account.objects.create(owner='alice', balance=100)
        self.bob = Account.objects.create(owner='bob', balance=20)

    def test_transfer_moves_money(self):
        transfer(self.alice.pk, self.bob.pk, 30)
        self.alice.refresh_from_db()
        self.bob.refresh_from_db()
        self.assertEqual((self.alice.balance, self.bob.balance), (70, 50))
        self.assertEqual(Transfer.objects.count(), 1)

    def test_insufficient_funds_changes_nothing(self):
        with self.assertRaises(InsufficientFunds):
            transfer(self.bob.pk, self.alice.pk, 50)
        self.bob.refresh_from_db()
        self.assertEqual(self.bob.balance, 20)
        self.assertEqual(Transfer.objects.count(), 0)

    def test_missing_target_rolls_back_the_debit(self):
        with self.assertRaises(Account.DoesNotExist):
            transfer(self.alice.pk, 999, 30)
        self.alice.refresh_from_db()
        self.assertEqual(self.alice.balance, 100)

    def test_check_constraint(self):
        with self.assertRaises(IntegrityError), transaction.atomic():
            Account.objects.filter(pk=self.bob.pk).update(balance=-1)

    def test_on_commit_callbacks(self):
        calls = []
        with self.captureOnCommitCallbacks(execute=True) as callbacks:
            with transaction.atomic():
                transaction.on_commit(lambda: calls.append('sent'))
                self.assertEqual(calls, [])
        self.assertEqual(len(callbacks), 1)
        self.assertEqual(calls, ['sent'])

The tests confirm that a failed transfer leaves the balances and the table of transfers untouched. The check constraint test wraps the failing statement in its own atomic block: after a database error the surrounding transaction is unusable until it is rolled back to a savepoint, and the test's own transaction would otherwise be broken.

$ uv run python manage.py test
Found 5 test(s).
System check identified no issues (0 silenced).
Creating test database for alias 'default'...
.....
----------------------------------------------------------------------
Ran 5 tests in 0.005s

OK
Destroying test database for alias 'default'...

Source

Django documentation: Performing raw SQL queries

Django documentation: Database transactions

Django documentation: Databases

Django documentation: select_for_update()

In this article we have run raw SQL with raw() and a cursor, passed parameters instead of formatting strings, grouped changes with transaction.atomic() and savepoints, delayed side effects with on_commit(), and used ATOMIC_REQUESTS. Related articles are Django query performance and Django QuerySets. Next we look at managers and custom QuerySets.

Author

My name is Jan Bodnar, and I am a passionate programmer with extensive programming experience. I have been writing programming articles since 2007. To date, I have authored over 1,400 articles and 8 e-books. I possess more than ten years of experience in teaching programming.

List all Django tutorials.