INTPYTHON-355 Add transaction support

Author: timgraham

.github/workflows/test-python-atlas.yml

@@ -53,4 +53,4 @@ jobs:
  working-directory: .
  run: bash .github/workflows/start_local_atlas.sh mongodb/mongodb-atlas-local:7
  - name: Run tests
- run: python3 django_repo/tests/runtests_.py
+ run: python3 django_repo/tests/runtests.py --settings mongodb_settings -v 2

django_mongodb_backend/base.py

@@ -2,7 +2,9 @@
 import os
 
 from django.core.exceptions import ImproperlyConfigured
+from django.db import DEFAULT_DB_ALIAS
 from django.db.backends.base.base import BaseDatabaseWrapper
+from django.db.backends.utils import debug_transaction
 from django.utils.asyncio import async_unsafe
 from django.utils.functional import cached_property
 from pymongo.collection import Collection
@@ -32,6 +34,17 @@ def __exit__(self, exception_type, exception_value, exception_traceback):
  pass
 
 
+def requires_transaction_support(func):
+ """Make a method a no-op if transactions aren't supported."""
+
+ def wrapper(self, *args, **kwargs):
+ if not self.features.supports_transactions:
+ return
+ func(self, *args, **kwargs)
+
+ return wrapper
+
+
 class DatabaseWrapper(BaseDatabaseWrapper):
  data_types = {
  "AutoField": "int",
@@ -140,6 +153,10 @@ def _isnull_operator(a, b):
  ops_class = DatabaseOperations
  validation_class = DatabaseValidation
 
+ def __init__(self, settings_dict, alias=DEFAULT_DB_ALIAS):
+ super().__init__(settings_dict, alias=alias)
+ self.session = None
+
  def get_collection(self, name, **kwargs):
  collection = Collection(self.database, name, **kwargs)
  if self.queries_logged:
@@ -189,14 +206,48 @@ def _driver_info(self):
  return DriverInfo("django-mongodb-backend", django_mongodb_backend_version)
  return None
 
+ @requires_transaction_support
  def _commit(self):
- pass
+ if self.session:
+ with debug_transaction(self, "session.commit_transaction()"):
+ self.session.commit_transaction()
+ self._end_session()
 
+ @requires_transaction_support
  def _rollback(self):
- pass
+ if self.session:
+ with debug_transaction(self, "session.abort_transaction()"):
+ self.session.abort_transaction()
+ self._end_session()
+
+ def _start_transaction(self):
+ # Private API, specific to this backend.
+ if self.session is None:
+ self.session = self.connection.start_session()
+ with debug_transaction(self, "session.start_transaction()"):
+ self.session.start_transaction()
 
- def set_autocommit(self, autocommit, force_begin_transaction_with_broken_autocommit=False):
- self.autocommit = autocommit
+ def _end_session(self):
+ # Private API, specific to this backend.
+ self.session.end_session()
+ self.session = None
+
+ @requires_transaction_support
+ def _start_transaction_under_autocommit(self):
+ # Implementing this hook (intended only for SQLite), allows
+ # BaseDatabaseWrapper.set_autocommit() to use it to start a transaction
+ # rather than set_autocommit(), bypassing set_autocommit()'s call to
+ # debug_transaction(self, "BEGIN") which isn't semantic for a no-SQL
+ # backend.
+ self._start_transaction()
+
+ @requires_transaction_support
+ def _set_autocommit(self, autocommit, force_begin_transaction_with_broken_autocommit=False):
+ # Besides @transaction.atomic() (which uses
+ # _start_transaction_under_autocommit(), disabling autocommit is
+ # another way to start a transaction.
+ if not autocommit:
+ self._start_transaction()
 
  def _close(self):
  # Normally called by close(), this method is also called by some tests.
@@ -210,6 +261,10 @@ def close(self):
 
  def close_pool(self):
  """Close the MongoClient."""
+ # Clear commit hooks and session.
+ self.run_on_commit = []
+ if self.session:
+ self._end_session()
  connection = self.connection
  if connection is None:
  return
@@ -225,6 +280,10 @@ def close_pool(self):
  def cursor(self):
  return Cursor()
 
+ @requires_transaction_support
+ def validate_no_broken_transaction(self):
+ super().validate_no_broken_transaction()
+
  def get_database_version(self):
  """Return a tuple of the database's version."""
  return tuple(self.connection.server_info()["versionArray"])

django_mongodb_backend/compiler.py

@@ -685,7 +685,10 @@ def execute_sql(self, returning_fields=None):
  @wrap_database_errors
  def insert(self, docs, returning_fields=None):
  """Store a list of documents using field columns as element names."""
- inserted_ids = self.collection.insert_many(docs).inserted_ids
+ self.connection.validate_no_broken_transaction()
+ inserted_ids = self.collection.insert_many(
+ docs, session=self.connection.session
+ ).inserted_ids
  return [(x,) for x in inserted_ids] if returning_fields else []
 
  @cached_property
@@ -768,7 +771,10 @@ def execute_sql(self, result_type):
 
  @wrap_database_errors
  def update(self, criteria, pipeline):
- return self.collection.update_many(criteria, pipeline).matched_count
+ self.connection.validate_no_broken_transaction()
+ return self.collection.update_many(
+ criteria, pipeline, session=self.connection.session
+ ).matched_count
 
  def check_query(self):
  super().check_query()

django_mongodb_backend/features.py

@@ -36,8 +36,6 @@ class DatabaseFeatures(BaseDatabaseFeatures):
  supports_temporal_subtraction = True
  # MongoDB stores datetimes in UTC.
  supports_timezones = False
- # Not implemented: https://github.com/mongodb/django-mongodb-backend/issues/7
- supports_transactions = False
  supports_unspecified_pk = True
  uses_savepoints = False
 
@@ -50,8 +48,6 @@ class DatabaseFeatures(BaseDatabaseFeatures):
  "aggregation.tests.AggregateTestCase.test_order_by_aggregate_transform",
  # 'NulledTransform' object has no attribute 'as_mql'.
  "lookup.tests.LookupTests.test_exact_none_transform",
- # "Save with update_fields did not affect any rows."
- "basic.tests.SelectOnSaveTests.test_select_on_save_lying_update",
  # BaseExpression.convert_value() crashes with Decimal128.
  "aggregation.tests.AggregateTestCase.test_combine_different_types",
  "annotations.tests.NonAggregateAnnotationTestCase.test_combined_f_expression_annotation_with_aggregation",
@@ -96,13 +92,47 @@ class DatabaseFeatures(BaseDatabaseFeatures):
  "expressions.tests.ExpressionOperatorTests.test_lefthand_bitwise_xor_right_null",
  "expressions.tests.ExpressionOperatorTests.test_lefthand_transformed_field_bitwise_or",
  }
+ _django_test_expected_failures_no_transactions = {
+ # "Save with update_fields did not affect any rows." instead of
+ # "An error occurred in the current transaction. You can't execute
+ # queries until the end of the 'atomic' block."
+ "basic.tests.SelectOnSaveTests.test_select_on_save_lying_update",
+ }
+ _django_test_expected_failures_transactions = {
+ # When update_or_create() fails with IntegrityError, the transaction
+ # is no longer usable.
+ "get_or_create.tests.UpdateOrCreateTests.test_manual_primary_key_test",
+ "get_or_create.tests.UpdateOrCreateTestsWithManualPKs.test_create_with_duplicate_primary_key",
+ # Tests that require savepoints
+ "admin_views.tests.AdminViewBasicTest.test_disallowed_to_field",
+ "admin_views.tests.AdminViewPermissionsTest.test_add_view",
+ "admin_views.tests.AdminViewPermissionsTest.test_change_view",
+ "admin_views.tests.AdminViewPermissionsTest.test_change_view_save_as_new",
+ "admin_views.tests.AdminViewPermissionsTest.test_delete_view",
+ "auth_tests.test_views.ChangelistTests.test_view_user_password_is_readonly",
+ "fixtures.tests.FixtureLoadingTests.test_loaddata_app_option",
+ "fixtures.tests.FixtureLoadingTests.test_unmatched_identifier_loading",
+ "fixtures_model_package.tests.FixtureTestCase.test_loaddata",
+ "get_or_create.tests.GetOrCreateTests.test_get_or_create_invalid_params",
+ "get_or_create.tests.UpdateOrCreateTests.test_integrity",
+ "many_to_many.tests.ManyToManyTests.test_add",
+ "many_to_one.tests.ManyToOneTests.test_fk_assignment_and_related_object_cache",
+ "model_fields.test_booleanfield.BooleanFieldTests.test_null_default",
+ "model_fields.test_floatfield.TestFloatField.test_float_validates_object",
+ "multiple_database.tests.QueryTestCase.test_generic_key_cross_database_protection",
+ "multiple_database.tests.QueryTestCase.test_m2m_cross_database_protection",
+ }
 
  @cached_property
  def django_test_expected_failures(self):
  expected_failures = super().django_test_expected_failures
  expected_failures.update(self._django_test_expected_failures)
  if not self.is_mongodb_6_3:
  expected_failures.update(self._django_test_expected_failures_bitwise)
+ if self.supports_transactions:
+ expected_failures.update(self._django_test_expected_failures_transactions)
+ else:
+ expected_failures.update(self._django_test_expected_failures_no_transactions)
  return expected_failures
 
  django_test_skips = {
@@ -485,16 +515,6 @@ def django_test_expected_failures(self):
  "Connection health checks not implemented.": {
  "backends.base.test_base.ConnectionHealthChecksTests",
  },
- "transaction.atomic() is not supported.": {
- "backends.base.test_base.DatabaseWrapperLoggingTests",
- "migrations.test_executor.ExecutorTests.test_atomic_operation_in_non_atomic_migration",
- "migrations.test_operations.OperationTests.test_run_python_atomic",
- },
- "transaction.rollback() is not supported.": {
- "transactions.tests.AtomicMiscTests.test_mark_for_rollback_on_error_in_autocommit",
- "transactions.tests.AtomicMiscTests.test_mark_for_rollback_on_error_in_transaction",
- "transactions.tests.NonAutocommitTests.test_orm_query_after_error_and_rollback",
- },
  "migrate --fake-initial is not supported.": {
  "migrations.test_commands.MigrateTests.test_migrate_fake_initial",
  "migrations.test_commands.MigrateTests.test_migrate_fake_split_initial",
@@ -587,3 +607,20 @@ def supports_atlas_search(self):
  return False
  else:
  return True
+
+ @cached_property
+ def supports_select_union(self):
+ # Stage not supported inside of a multi-document transaction: $unionWith
+ return not self.supports_transactions
+
+ @cached_property
+ def supports_transactions(self):
+ """
+ Transactions are enabled if MongoDB is configured as a replica set or a
+ sharded cluster.
+ """
+ self.connection.ensure_connection()
+ client = self.connection.connection.admin
+ hello = client.command("hello")
+ # a replica set or a sharded cluster
+ return "setName" in hello or hello.get("msg") == "isdbgrid"

django_mongodb_backend/query.py

@@ -61,17 +61,23 @@ def __repr__(self):
  @wrap_database_errors
  def delete(self):
  """Execute a delete query."""
+ self.compiler.connection.validate_no_broken_transaction()
  if self.compiler.subqueries:
  raise NotSupportedError("Cannot use QuerySet.delete() when a subquery is required.")
- return self.compiler.collection.delete_many(self.match_mql).deleted_count
+ return self.compiler.collection.delete_many(
+ self.match_mql, session=self.compiler.connection.session
+ ).deleted_count
 
  @wrap_database_errors
  def get_cursor(self):
  """
  Return a pymongo CommandCursor that can be iterated on to give the
  results of the query.
  """
- return self.compiler.collection.aggregate(self.get_pipeline())
+ self.compiler.connection.validate_no_broken_transaction()
+ return self.compiler.collection.aggregate(
+ self.get_pipeline(), session=self.compiler.connection.session
+ )
 
  def get_pipeline(self):
  pipeline = []

django_mongodb_backend/queryset.py

@@ -35,7 +35,7 @@ def __init__(self, pipeline, using, model):
  def _execute_query(self):
  connection = connections[self.using]
  collection = connection.get_collection(self.model._meta.db_table)
- self.cursor = collection.aggregate(self.pipeline)
+ self.cursor = collection.aggregate(self.pipeline, session=connection.session)
 
  def __str__(self):
  return str(self.pipeline)

docs/source/conf.py

@@ -45,6 +45,7 @@
  "pymongo": ("https://pymongo.readthedocs.io/en/stable/", None),
  "python": ("https://docs.python.org/3/", None),
  "atlas": ("https://www.mongodb.com/docs/atlas/", None),
+ "manual": ("https://www.mongodb.com/docs/manual/", None),
 }
 
 root_doc = "contents"

docs/source/ref/database.rst

@@ -41,3 +41,43 @@ effect. Rather, if you need to close the connection pool, use
 .. versionadded:: 5.2.0b0
 
  Support for connection pooling and ``connection.close_pool()`` were added.
+
+.. _transactions:
+
+Transactions
+============
+
+.. versionadded:: 5.2.0b2
+
+Support for :doc:`Django's transactions APIs <django:topics/db/transactions>`
+is enabled if MongoDB is configured as a :doc:`replica set<manual:replication>`
+or a :doc:`sharded cluster <manual:sharding>`.
+
+If transactions aren't supported, query execution uses Django and MongoDB's
+default behavior of autocommit mode. Each query is immediately committed to the
+database. Django's transaction management APIs, such as
+:func:`~django.db.transaction.atomic`, function as no-ops.
+
+.. _transactions-limitations:
+
+Limitations
+-----------
+
+MongoDB's transaction limitations that are applicable to Django are:
+
+- :meth:`QuerySet.union() <django.db.models.query.QuerySet.union>` is not
+ supported inside a transaction.
+- If a transaction raises an exception, the transaction is no longer usable.
+ For example, if the update stage of :meth:`QuerySet.update_or_create()
+ <django.db.models.query.QuerySet.update_or_create>` fails with
+ :class:`~django.db.IntegrityError` due to a unique constraint violation, the
+ create stage won't be able to proceed.
+ :class:`pymongo.errors.OperationFailure` is raised, wrapped by
+ :class:`django.db.DatabaseError`.
+- Savepoints (i.e. nested :func:`~django.db.transaction.atomic` blocks) aren't
+ supported. The outermost :func:`~django.db.transaction.atomic` will start
+ a transaction while any subsequent :func:`~django.db.transaction.atomic`
+ blocks will have no effect.
+- Migration operations aren't :ref:`wrapped in a transaction
+ <topics/migrations:transactions>` because of MongoDB restrictions such as
+ adding indexes to existing collections while in a transaction.

docs/source/releases/5.2.x.rst

@@ -13,6 +13,7 @@ New features
 - Added subquery support for :class:`~.fields.EmbeddedModelArrayField`.
 - Added the ``options`` parameter to
  :func:`~django_mongodb_backend.utils.parse_uri`.
+- Added support for :ref:`database transactions <transactions>`.
 
 5.2.0 beta 1
 ============

docs/source/topics/known-issues.rst

@@ -80,11 +80,7 @@ Database functions
 Transaction management
 ======================
 
-Query execution uses Django and MongoDB's default behavior of autocommit mode.
-Each query is immediately committed to the database.
-
-Django's :doc:`transaction management APIs <django:topics/db/transactions>`
-are not supported.
+See :ref:`transactions` for details.
 
 Database introspection
 ======================