PYTHON-5981 Reject aggregate/pipeline as aggregation options

doc/changelog.rst

@@ -39,6 +39,18 @@ PyMongo 4.18 brings a number of changes including:
 - Fixed a bug on Windows, and on macOS when using PyOpenSSL, where
   ``SSL_CERT_FILE``/``SSL_CERT_DIR`` were merged with, rather than replacing,
   the OS/certifi certificate store.
+- Aggregation helpers now raise :exc:`~pymongo.errors.ConfigurationError` when
+  passed an ``aggregate`` or ``pipeline`` keyword argument. Previously these
+  keys silently replaced the target namespace and pipeline of the generated
+  ``aggregate`` command. This affects
+  :meth:`~pymongo.asynchronous.collection.AsyncCollection.aggregate` and
+  :meth:`~pymongo.synchronous.collection.Collection.aggregate`,
+  :meth:`~pymongo.asynchronous.collection.AsyncCollection.aggregate_raw_batches`
+  and :meth:`~pymongo.synchronous.collection.Collection.aggregate_raw_batches`,
+  :meth:`~pymongo.asynchronous.database.AsyncDatabase.aggregate` and
+  :meth:`~pymongo.synchronous.database.Database.aggregate`, and
+  :meth:`~pymongo.asynchronous.collection.AsyncCollection.list_search_indexes`
+  and :meth:`~pymongo.synchronous.collection.Collection.list_search_indexes`.
 
 Changes in Version 4.17.0 (2026/04/20)
 --------------------------------------

pymongo/asynchronous/aggregation.py

@@ -60,6 +60,11 @@ def __init__(
             raise ConfigurationError(
                 "The explain option is not supported. Use AsyncDatabase.command instead."
             )
+        for name in ("aggregate", "pipeline"):
+            if name in options:
+                raise ConfigurationError(
+                    f"The {name} option cannot be specified as a keyword argument."
+                )
 
         self._target = target
 

pymongo/synchronous/aggregation.py

@@ -60,6 +60,11 @@ def __init__(
             raise ConfigurationError(
                 "The explain option is not supported. Use Database.command instead."
             )
+        for name in ("aggregate", "pipeline"):
+            if name in options:
+                raise ConfigurationError(
+                    f"The {name} option cannot be specified as a keyword argument."
+                )
 
         self._target = target
 

test/asynchronous/test_collection.py

@@ -1568,6 +1568,30 @@ async def test_aggregate(self):
         with self.write_concern_collection() as coll:
             await coll.aggregate([{"$out": "output-collection"}])
 
+    async def test_aggregate_reserved_options(self):
+        # "aggregate" and "pipeline" are fields of the aggregate command itself,
+        # so they must not be settable as keyword options: doing so would
+        # replace the command's target namespace or its pipeline.
+        db = self.db
+        reserved_options: list[dict[str, Any]] = [
+            {"aggregate": "other"},
+            {"pipeline": [{"$out": "other"}]},
+            {"aggregate": "other", "pipeline": [{"$out": "other"}]},
+        ]
+        for options in reserved_options:
+            with self.subTest(options=options):
+                # These helpers take the pipeline positionally, so only pass
+                # the options that do not collide with it.
+                if "pipeline" not in options:
+                    with self.assertRaises(ConfigurationError):
+                        await db.test.aggregate([], **options)
+                    with self.assertRaises(ConfigurationError):
+                        await db.test.aggregate_raw_batches([], **options)
+                    with self.assertRaises(ConfigurationError):
+                        await db.aggregate([], **options)
+                with self.assertRaises(ConfigurationError):
+                    await db.test.list_search_indexes(**options)
+
     async def test_aggregate_raw_bson(self):
         db = self.db
         await db.drop_collection("test")

test/test_collection.py

@@ -1550,6 +1550,30 @@ def test_aggregate(self):
         with self.write_concern_collection() as coll:
             coll.aggregate([{"$out": "output-collection"}])
 
+    def test_aggregate_reserved_options(self):
+        # "aggregate" and "pipeline" are fields of the aggregate command itself,
+        # so they must not be settable as keyword options: doing so would
+        # replace the command's target namespace or its pipeline.
+        db = self.db
+        reserved_options: list[dict[str, Any]] = [
+            {"aggregate": "other"},
+            {"pipeline": [{"$out": "other"}]},
+            {"aggregate": "other", "pipeline": [{"$out": "other"}]},
+        ]
+        for options in reserved_options:
+            with self.subTest(options=options):
+                # These helpers take the pipeline positionally, so only pass
+                # the options that do not collide with it.
+                if "pipeline" not in options:
+                    with self.assertRaises(ConfigurationError):
+                        db.test.aggregate([], **options)
+                    with self.assertRaises(ConfigurationError):
+                        db.test.aggregate_raw_batches([], **options)
+                    with self.assertRaises(ConfigurationError):
+                        db.aggregate([], **options)
+                with self.assertRaises(ConfigurationError):
+                    db.test.list_search_indexes(**options)
+
     def test_aggregate_raw_bson(self):
         db = self.db
         db.drop_collection("test")