diff --git a/.evergreen/run-tests.sh b/.evergreen/run-tests.sh
index f49a0e9a..85f3bd33 100644
--- a/.evergreen/run-tests.sh
+++ b/.evergreen/run-tests.sh
@@ -3,13 +3,13 @@
 set -eux
 
 # Install django-mongodb-backend
-/opt/python/3.10/bin/python3 -m venv venv
+/opt/python/3.12/bin/python3 -m venv venv
 . venv/bin/activate
 python -m pip install -U pip
 pip install -e .
 
 # Install django and test dependencies
-git clone --branch mongodb-5.2.x https://github.com/mongodb-forks/django django_repo
+git clone --branch mongodb-6.0.x https://github.com/mongodb-forks/django django_repo
 pushd django_repo/tests/
 pip install -e ..
 pip install -r requirements/py3.txt
diff --git a/.github/workflows/linters.yml b/.github/workflows/linters.yml
index dcf46d45..4bf0fa77 100644
--- a/.github/workflows/linters.yml
+++ b/.github/workflows/linters.yml
@@ -17,7 +17,7 @@ jobs:
           persist-credentials: false
       - uses: actions/setup-python@v5
         with:
-          python-version: '3.10'
+          python-version: '3.12'
           cache: 'pip'
           cache-dependency-path: 'pyproject.toml'
       - name: Install Python dependencies
@@ -37,7 +37,7 @@ jobs:
         with:
           cache: 'pip'
           cache-dependency-path: 'pyproject.toml'
-          python-version: '3.10'
+          python-version: '3.12'
       - name: Install dependencies
         run: |
           pip install -U pip
diff --git a/.github/workflows/test-python-atlas.yml b/.github/workflows/test-python-atlas.yml
index 175dfe18..5c1770b9 100644
--- a/.github/workflows/test-python-atlas.yml
+++ b/.github/workflows/test-python-atlas.yml
@@ -33,7 +33,7 @@ jobs:
         uses: actions/checkout@v4
         with:
           repository: 'mongodb-forks/django'
-          ref: 'mongodb-5.2.x'
+          ref: 'mongodb-6.0.x'
           path: 'django_repo'
           persist-credentials: false
       - name: Install system packages for Django's Python test dependencies
@@ -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
diff --git a/.github/workflows/test-python.yml b/.github/workflows/test-python.yml
index d3fccebd..5f912bba 100644
--- a/.github/workflows/test-python.yml
+++ b/.github/workflows/test-python.yml
@@ -33,7 +33,7 @@ jobs:
         uses: actions/checkout@v4
         with:
           repository: 'mongodb-forks/django'
-          ref: 'mongodb-5.2.x'
+          ref: 'mongodb-6.0.x'
           path: 'django_repo'
           persist-credentials: false
       - name: Install system packages for Django's Python test dependencies
diff --git a/.readthedocs.yaml b/.readthedocs.yaml
index bf5dba2f..ba11945f 100644
--- a/.readthedocs.yaml
+++ b/.readthedocs.yaml
@@ -21,6 +21,6 @@ python:
           - docs
 
 build:
-  os: ubuntu-22.04
+  os: ubuntu-24.04
   tools:
-    python: "3.11"
+    python: "3.12"
diff --git a/django_mongodb_backend/__init__.py b/django_mongodb_backend/__init__.py
index 00700421..f57b3e7c 100644
--- a/django_mongodb_backend/__init__.py
+++ b/django_mongodb_backend/__init__.py
@@ -1,4 +1,4 @@
-__version__ = "5.2.0b2.dev0"
+__version__ = "6.0.0b0.dev0"
 
 # Check Django compatibility before other imports which may fail if the
 # wrong version of Django is installed.
diff --git a/django_mongodb_backend/aggregates.py b/django_mongodb_backend/aggregates.py
index e67cefa2..93df6be0 100644
--- a/django_mongodb_backend/aggregates.py
+++ b/django_mongodb_backend/aggregates.py
@@ -1,5 +1,12 @@
-from django.db.models.aggregates import Aggregate, Count, StdDev, Variance
-from django.db.models.expressions import Case, Value, When
+from django.db.models.aggregates import (
+    Aggregate,
+    AggregateFilter,
+    Count,
+    StdDev,
+    StringAgg,
+    Variance,
+)
+from django.db.models.expressions import Case, Col, Value, When
 from django.db.models.lookups import IsNull
 
 from .query_utils import process_lhs
@@ -16,7 +23,11 @@ def aggregate(
     resolve_inner_expression=False,
     **extra_context,  # noqa: ARG001
 ):
-    if self.filter:
+    # TODO: isinstance(self.filter, Col) works around failure of
+    # aggregation.tests.AggregateTestCase.test_distinct_on_aggregate. Is this
+    # correct?
+    if self.filter is not None and not isinstance(self.filter, Col):
+        # Generate a CASE statement for this aggregate.
         node = self.copy()
         node.filter = None
         source_expressions = node.get_source_expressions()
@@ -31,6 +42,10 @@ def aggregate(
     return {f"${operator}": lhs_mql}
 
 
+def aggregate_filter(self, compiler, connection, **extra_context):
+    return self.condition.as_mql(compiler, connection, **extra_context)
+
+
 def count(self, compiler, connection, resolve_inner_expression=False, **extra_context):  # noqa: ARG001
     """
     When resolve_inner_expression=True, return the MQL that resolves as a
@@ -72,8 +87,16 @@ def stddev_variance(self, compiler, connection, **extra_context):
     return aggregate(self, compiler, connection, operator=operator, **extra_context)
 
 
+def string_agg(self, compiler, connection, **extra_context):  # # noqa: ARG001
+    from django.db import NotSupportedError
+
+    raise NotSupportedError("StringAgg is not supported.")
+
+
 def register_aggregates():
     Aggregate.as_mql = aggregate
+    AggregateFilter.as_mql = aggregate_filter
     Count.as_mql = count
     StdDev.as_mql = stddev_variance
+    StringAgg.as_mql = string_agg
     Variance.as_mql = stddev_variance
diff --git a/django_mongodb_backend/base.py b/django_mongodb_backend/base.py
index c6110dbb..fc21fa5b 100644
--- a/django_mongodb_backend/base.py
+++ b/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"])
diff --git a/django_mongodb_backend/cache.py b/django_mongodb_backend/cache.py
index 00b903af..ae4809a2 100644
--- a/django_mongodb_backend/cache.py
+++ b/django_mongodb_backend/cache.py
@@ -1,5 +1,5 @@
 import pickle
-from datetime import datetime, timezone
+from datetime import UTC, datetime
 
 from django.core.cache.backends.base import DEFAULT_TIMEOUT, BaseCache
 from django.core.cache.backends.db import Options
@@ -72,7 +72,7 @@ def get_backend_timeout(self, timeout=DEFAULT_TIMEOUT):
         if timeout is None:
             return datetime.max
         timestamp = super().get_backend_timeout(timeout)
-        return datetime.fromtimestamp(timestamp, tz=timezone.utc)
+        return datetime.fromtimestamp(timestamp, tz=UTC)
 
     def get(self, key, default=None, version=None):
         return self.get_many([key], version).get(key, default)
diff --git a/django_mongodb_backend/compiler.py b/django_mongodb_backend/compiler.py
index 12da13a1..b7e264f8 100644
--- a/django_mongodb_backend/compiler.py
+++ b/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()
diff --git a/django_mongodb_backend/features.py b/django_mongodb_backend/features.py
index fa73461d..b5127d8a 100644
--- a/django_mongodb_backend/features.py
+++ b/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",
@@ -86,6 +82,21 @@ class DatabaseFeatures(BaseDatabaseFeatures):
         # Value.as_mql() doesn't call output_field.get_db_prep_save():
         # https://github.com/mongodb/django-mongodb-backend/issues/282
         "model_fields.test_jsonfield.TestSaveLoad.test_bulk_update_custom_get_prep_value",
+        # This backend overrides DatabaseCreation.create_test_db() so the
+        # deprecation warnings stacklevel points to the wrong file.
+        "backends.base.test_creation.TestDbCreationTests.test_serialize_deprecation",
+        # StringAgg is not supported.
+        "aggregation.tests.AggregateTestCase.test_distinct_on_stringagg",
+        "aggregation.tests.AggregateTestCase.test_string_agg_escapes_delimiter",
+        "aggregation.tests.AggregateTestCase.test_string_agg_filter",
+        "aggregation.tests.AggregateTestCase.test_string_agg_filter_in_subquery",
+        "aggregation.tests.AggregateTestCase.test_stringagg_default_value",
+        # bulk_create() population of _order not implemented.
+        # https://github.com/django/django/commit/953095d1e603fe0f8f01175b1409ca23818dcff9
+        "contenttypes_tests.test_order_with_respect_to.OrderWithRespectToGFKTests.test_bulk_create_allows_duplicate_order_values",
+        "contenttypes_tests.test_order_with_respect_to.OrderWithRespectToGFKTests.test_bulk_create_mixed_scenario",
+        "contenttypes_tests.test_order_with_respect_to.OrderWithRespectToGFKTests.test_bulk_create_respects_mixed_manual_order",
+        "contenttypes_tests.test_order_with_respect_to.OrderWithRespectToGFKTests.test_bulk_create_with_existing_children",
     }
     # $bitAnd, #bitOr, and $bitXor are new in MongoDB 6.3.
     _django_test_expected_failures_bitwise = {
@@ -96,6 +107,36 @@ 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):
@@ -103,6 +144,10 @@ def django_test_expected_failures(self):
         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 +530,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",
@@ -533,8 +568,18 @@ def django_test_expected_failures(self):
             "foreign_object.test_tuple_lookups.TupleLookupsTests",
         },
         "ColPairs is not supported.": {
-            # 'ColPairs' object has no attribute 'as_mql'
             "auth_tests.test_views.CustomUserCompositePrimaryKeyPasswordResetTest",
+            "composite_pk.test_aggregate.CompositePKAggregateTests",
+            "composite_pk.test_create.CompositePKCreateTests",
+            "composite_pk.test_delete.CompositePKDeleteTests",
+            "composite_pk.test_filter.CompositePKFilterTests",
+            "composite_pk.test_get.CompositePKGetTests",
+            "composite_pk.test_models.CompositePKModelsTests",
+            "composite_pk.test_order_by.CompositePKOrderByTests",
+            "composite_pk.test_update.CompositePKUpdateTests",
+            "composite_pk.test_values.CompositePKValuesTests",
+            "composite_pk.tests.CompositePKTests",
+            "composite_pk.tests.CompositePKFixturesTests",
         },
         "Custom lookups are not supported.": {
             "custom_lookups.tests.BilateralTransformTests",
@@ -577,3 +622,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"
diff --git a/django_mongodb_backend/fields/json.py b/django_mongodb_backend/fields/json.py
index b7cf49dc..24c640b4 100644
--- a/django_mongodb_backend/fields/json.py
+++ b/django_mongodb_backend/fields/json.py
@@ -24,7 +24,12 @@ def build_json_mql_path(lhs, key_transforms):
         get_field = {"$getField": {"input": result, "field": key}}
         # Handle array indexing if the key is a digit. If key is something
         # like '001', it's not an array index despite isdigit() returning True.
-        if key.isdigit() and str(int(key)) == key:
+        try:
+            int(key)
+            is_digit = str(int(key)) == key
+        except ValueError:
+            is_digit = False
+        if is_digit:
             result = {
                 "$cond": {
                     "if": {"$isArray": result},
diff --git a/django_mongodb_backend/query.py b/django_mongodb_backend/query.py
index 04977520..d59bc163 100644
--- a/django_mongodb_backend/query.py
+++ b/django_mongodb_backend/query.py
@@ -61,9 +61,12 @@ 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):
@@ -71,7 +74,10 @@ 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 = []
diff --git a/django_mongodb_backend/queryset.py b/django_mongodb_backend/queryset.py
index 4a2d884d..b02496fe 100644
--- a/django_mongodb_backend/queryset.py
+++ b/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)
diff --git a/docs/source/conf.py b/docs/source/conf.py
index a4e54938..2f1c8675 100644
--- a/docs/source/conf.py
+++ b/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"
diff --git a/docs/source/ref/database.rst b/docs/source/ref/database.rst
index 365e1c73..0abd3637 100644
--- a/docs/source/ref/database.rst
+++ b/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.
diff --git a/docs/source/releases/5.2.x.rst b/docs/source/releases/5.2.x.rst
index 6b17b86e..e2ab337a 100644
--- a/docs/source/releases/5.2.x.rst
+++ b/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
 ============
diff --git a/docs/source/topics/known-issues.rst b/docs/source/topics/known-issues.rst
index 4b9edee7..60628f81 100644
--- a/docs/source/topics/known-issues.rst
+++ b/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
 ======================
diff --git a/pyproject.toml b/pyproject.toml
index 7b412e1d..8f100a98 100644
--- a/pyproject.toml
+++ b/pyproject.toml
@@ -8,7 +8,7 @@ dynamic = ["version", "dependencies"]
 description = "Django MongoDB Backend"
 readme = "README.md"
 license = {file="LICENSE"}
-requires-python = ">=3.10"
+requires-python = ">=3.12"
 authors = [
     { name = "The MongoDB Python Team" },
 ]
@@ -27,8 +27,6 @@ classifiers = [
     "Operating System :: OS Independent",
     "Programming Language :: Python :: 3",
     "Programming Language :: Python :: 3 :: Only",
-    "Programming Language :: Python :: 3.10",
-    "Programming Language :: Python :: 3.11",
     "Programming Language :: Python :: 3.12",
     "Programming Language :: Python :: 3.13",
 ]
diff --git a/tests/backend_/test_base.py b/tests/backend_/test_base.py
index 7695b6f4..9d4e7006 100644
--- a/tests/backend_/test_base.py
+++ b/tests/backend_/test_base.py
@@ -1,7 +1,7 @@
 from django.core.exceptions import ImproperlyConfigured
 from django.db import connection
 from django.db.backends.signals import connection_created
-from django.test import SimpleTestCase, TestCase
+from django.test import SimpleTestCase, TransactionTestCase
 
 from django_mongodb_backend.base import DatabaseWrapper
 
@@ -15,7 +15,9 @@ def test_database_name_empty(self):
             DatabaseWrapper(settings).get_connection_params()
 
 
-class DatabaseWrapperConnectionTests(TestCase):
+class DatabaseWrapperConnectionTests(TransactionTestCase):
+    available_apps = ["backend_"]
+
     def test_set_autocommit(self):
         self.assertIs(connection.get_autocommit(), True)
         connection.set_autocommit(False)
diff --git a/tests/backend_/test_features.py b/tests/backend_/test_features.py
new file mode 100644
index 00000000..504d3a3f
--- /dev/null
+++ b/tests/backend_/test_features.py
@@ -0,0 +1,46 @@
+from unittest.mock import patch
+
+from django.db import connection
+from django.test import TestCase
+
+
+class SupportsTransactionsTests(TestCase):
+    def setUp(self):
+        # Clear the cached property.
+        del connection.features.supports_transactions
+
+    def tearDown(self):
+        del connection.features.supports_transactions
+
+    def test_replica_set(self):
+        """A replica set supports transactions."""
+
+        def mocked_command(command):
+            if command == "hello":
+                return {"setName": "foo"}
+            raise Exception("Unexpected command")
+
+        with patch("pymongo.synchronous.database.Database.command", wraps=mocked_command):
+            self.assertIs(connection.features.supports_transactions, True)
+
+    def test_sharded_cluster(self):
+        """A sharded cluster supports transactions."""
+
+        def mocked_command(command):
+            if command == "hello":
+                return {"msg": "isdbgrid"}
+            raise Exception("Unexpected command")
+
+        with patch("pymongo.synchronous.database.Database.command", wraps=mocked_command):
+            self.assertIs(connection.features.supports_transactions, True)
+
+    def test_no_support(self):
+        """No support on a non-replica set, non-sharded cluster."""
+
+        def mocked_command(command):
+            if command == "hello":
+                return {}
+            raise Exception("Unexpected command")
+
+        with patch("pymongo.synchronous.database.Database.command", wraps=mocked_command):
+            self.assertIs(connection.features.supports_transactions, False)
diff --git a/tests/queries_/test_objectid.py b/tests/queries_/test_objectid.py
index 490d1b33..36d9561a 100644
--- a/tests/queries_/test_objectid.py
+++ b/tests/queries_/test_objectid.py
@@ -1,6 +1,6 @@
 from bson import ObjectId
 from django.core.exceptions import ValidationError
-from django.test import TestCase
+from django.test import TestCase, skipUnlessDBFeature
 
 from .models import Order, OrderItem, Tag
 
@@ -75,6 +75,7 @@ def test_filter_parent_by_children_values_obj(self):
         parent_qs = Tag.objects.filter(children__id__in=child_ids).distinct().order_by("name")
         self.assertSequenceEqual(parent_qs, [self.t1])
 
+    @skipUnlessDBFeature("supports_select_union")
     def test_filter_group_id_union_with_str(self):
         """Combine queries using union with string values."""
         qs_a = Tag.objects.filter(group_id=self.group_id_str_1)
@@ -82,6 +83,7 @@ def test_filter_group_id_union_with_str(self):
         union_qs = qs_a.union(qs_b).order_by("name")
         self.assertSequenceEqual(union_qs, [self.t3, self.t4])
 
+    @skipUnlessDBFeature("supports_select_union")
     def test_filter_group_id_union_with_obj(self):
         """Combine queries using union with ObjectId values."""
         qs_a = Tag.objects.filter(group_id=self.group_id_obj_1)
diff --git a/tests/raw_query_/test_raw_aggregate.py b/tests/raw_query_/test_raw_aggregate.py
index 72cd74d0..ce87311a 100644
--- a/tests/raw_query_/test_raw_aggregate.py
+++ b/tests/raw_query_/test_raw_aggregate.py
@@ -182,7 +182,8 @@ def test_different_db_key_order(self):
             {
                 field.name: getattr(author, field.name)
                 for field in reversed(Author._meta.concrete_fields)
-            }
+            },
+            session=connection.session,
         )
         query = []
         authors = Author.objects.all()