Fixed #29280 -- Made the transactions behavior configurable on SQLite.

AUTHORS

@@ -103,6 +103,7 @@ answer newbie questions, and generally made Django that much better:
     Antti Kaihola <http://djangopeople.net/akaihola/>
     Anubhav Joshi <anubhav9042@gmail.com>
     Anvesh Mishra <anveshgreat11@gmail.com>
+    Anže Pečar <anze@pecar.me>
     Aram Dulyan
     arien <regexbot@gmail.com>
     Armin Ronacher

django/db/backends/sqlite3/base.py

@@ -135,6 +135,8 @@ class DatabaseWrapper(BaseDatabaseWrapper):
         "iendswith": r"LIKE '%%' || UPPER({}) ESCAPE '\'",
     }
 
+    transaction_modes = frozenset(["DEFERRED", "EXCLUSIVE", "IMMEDIATE"])
+
     Database = Database
     SchemaEditorClass = DatabaseSchemaEditor
     # Classes instantiated in __init__().
@@ -171,6 +173,20 @@ class DatabaseWrapper(BaseDatabaseWrapper):
                 RuntimeWarning,
             )
         kwargs.update({"check_same_thread": False, "uri": True})
+        transaction_mode = kwargs.pop("transaction_mode", None)
+        if (
+            transaction_mode is not None
+            and transaction_mode.upper() not in self.transaction_modes
+        ):
+            allowed_transaction_modes = ", ".join(
+                [f"{mode!r}" for mode in sorted(self.transaction_modes)]
+            )
+            raise ImproperlyConfigured(
+                f"settings.DATABASES[{self.alias!r}]['OPTIONS']['transaction_mode'] "
+                f"is improperly configured to '{transaction_mode}'. Use one of "
+                f"{allowed_transaction_modes}, or None."
+            )
+        self.transaction_mode = transaction_mode.upper() if transaction_mode else None
         return kwargs
 
     def get_database_version(self):
@@ -298,7 +314,10 @@ class DatabaseWrapper(BaseDatabaseWrapper):
         Staying in autocommit mode works around a bug of sqlite3 that breaks
         savepoints when autocommit is disabled.
         """
+        if self.transaction_mode is None:
             self.cursor().execute("BEGIN")
+        else:
+            self.cursor().execute(f"BEGIN {self.transaction_mode}")
 
     def is_in_memory_db(self):
         return self.creation.is_in_memory_db(self.settings_dict["NAME"])

docs/ref/databases.txt

@@ -870,6 +870,38 @@ If you're getting this error, you can solve it by:
   This will make SQLite wait a bit longer before throwing "database is locked"
   errors; it won't really do anything to solve them.
 
+.. _sqlite-transaction-behavior:
+
+Transactions behavior
+~~~~~~~~~~~~~~~~~~~~~
+
+.. versionadded:: 5.1
+
+SQLite supports three transaction modes: ``DEFERRED``, ``IMMEDIATE``, and
+``EXCLUSIVE``.
+
+The default is ``DEFERRED``. If you need to use a different mode, set it in the
+:setting:`OPTIONS` part of your database configuration in
+:setting:`DATABASES`, for example::
+
+    "OPTIONS": {
+        # ...
+        "transaction_mode": "IMMEDIATE",
+        # ...
+    }
+
+To make sure your transactions wait until ``timeout`` before raising "Database
+is Locked", change the transaction mode to ``IMMEDIATE``.
+
+For the best performance with ``IMMEDIATE`` and ``EXCLUSIVE``, transactions
+should be as short as possible. This might be hard to guarantee for all of your
+views so the usage of :setting:`ATOMIC_REQUESTS <DATABASE-ATOMIC_REQUESTS>` is
+discouraged  in this case.
+
+For more information see `Transactions in SQLite`_.
+
+.. _`Transactions in SQLite`:  https://www.sqlite.org/lang_transaction.html#deferred_immediate_and_exclusive_transactions
+
 ``QuerySet.select_for_update()`` not supported
 ----------------------------------------------
 

docs/releases/5.1.txt

@@ -235,6 +235,9 @@ Models
   reload a model's value. This can be used to lock the row before reloading or
   to select related objects.
 
+* The new ``"transaction_mode"`` option is now supported in :setting:`OPTIONS`
+  on SQLite to allow specifying the :ref:`sqlite-transaction-behavior`.
+
 Requests and Responses
 ~~~~~~~~~~~~~~~~~~~~~~
 

tests/backends/sqlite/tests.py

@@ -3,9 +3,11 @@ import re
 import tempfile
 import threading
 import unittest
+from contextlib import contextmanager
 from pathlib import Path
 from unittest import mock
 
+from django.core.exceptions import ImproperlyConfigured
 from django.db import (
     DEFAULT_DB_ALIAS,
     NotSupportedError,
@@ -15,8 +17,8 @@ from django.db import (
 )
 from django.db.models import Aggregate, Avg, StdDev, Sum, Variance
 from django.db.utils import ConnectionHandler
-from django.test import TestCase, TransactionTestCase, override_settings
+from django.test import SimpleTestCase, TestCase, TransactionTestCase, override_settings
-from django.test.utils import isolate_apps
+from django.test.utils import CaptureQueriesContext, isolate_apps
 
 from ..models import Item, Object, Square
 
@@ -245,3 +247,55 @@ class ThreadSharing(TransactionTestCase):
             for conn in thread_connections:
                 if conn is not main_connection:
                     conn.close()
+
+
+@unittest.skipUnless(connection.vendor == "sqlite", "SQLite tests")
+class TestTransactionMode(SimpleTestCase):
+    databases = {"default"}
+
+    def test_default_transaction_mode(self):
+        with CaptureQueriesContext(connection) as captured_queries:
+            with transaction.atomic():
+                pass
+
+        begin_query, commit_query = captured_queries
+        self.assertEqual(begin_query["sql"], "BEGIN")
+        self.assertEqual(commit_query["sql"], "COMMIT")
+
+    def test_invalid_transaction_mode(self):
+        msg = (
+            "settings.DATABASES['default']['OPTIONS']['transaction_mode'] is "
+            "improperly configured to 'invalid'. Use one of 'DEFERRED', 'EXCLUSIVE', "
+            "'IMMEDIATE', or None."
+        )
+        with self.change_transaction_mode("invalid") as new_connection:
+            with self.assertRaisesMessage(ImproperlyConfigured, msg):
+                new_connection.ensure_connection()
+
+    def test_valid_transaction_modes(self):
+        valid_transaction_modes = ("deferred", "immediate", "exclusive")
+        for transaction_mode in valid_transaction_modes:
+            with (
+                self.subTest(transaction_mode=transaction_mode),
+                self.change_transaction_mode(transaction_mode) as new_connection,
+                CaptureQueriesContext(new_connection) as captured_queries,
+            ):
+                new_connection.set_autocommit(
+                    False, force_begin_transaction_with_broken_autocommit=True
+                )
+                new_connection.commit()
+                expected_transaction_mode = transaction_mode.upper()
+                begin_sql = captured_queries[0]["sql"]
+                self.assertEqual(begin_sql, f"BEGIN {expected_transaction_mode}")
+
+    @contextmanager
+    def change_transaction_mode(self, transaction_mode):
+        new_connection = connection.copy()
+        new_connection.settings_dict["OPTIONS"] = {
+            **new_connection.settings_dict["OPTIONS"],
+            "transaction_mode": transaction_mode,
+        }
+        try:
+            yield new_connection
+        finally:
+            new_connection.close()