MongoDB Adapter: Evaluate filtering by _id by casting to bson.ObjectId

Native objects can't be passed using JSON, so this is a minor surrogate
representation to convey a filter that makes `ctk load table` process
a single document, identified by its OID.

CHANGES.md

@@ -11,6 +11,9 @@
 - MongoDB Adapter: Improved performance of when computing collection cardinality
   by using `collection.estimated_document_count()`
 - MongoDB Full: Optionally use `limit` parameter as number of total records
+- MongoDB Adapter: Evaluate `_id` filter field by upcasting to `bson.ObjectId`,
+  to convey a filter that makes `ctk load table` process a single document,
+  identified by its OID
 
 ## 2024/09/19 v0.0.24
 - MongoDB Full: Refactor transformation subsystem to `commons-codec`

cratedb_toolkit/io/mongodb/adapter.py

@@ -4,6 +4,7 @@
 import logging
 import typing as t
 from abc import abstractmethod
+from copy import deepcopy
 from functools import cached_property
 from pathlib import Path
 
@@ -57,7 +58,7 @@ def batch_size(self) -> int:
         return int(self.address.uri.query_params.get("batch-size", 100))
 
     @cached_property
-    def filter(self) -> t.Union[str, None]:
+    def filter(self) -> t.Union[t.Dict[str, t.Any], None]:
         return json.loads(self.address.uri.query_params.get("filter", "null"))
 
     @cached_property
@@ -179,8 +180,12 @@ def record_count(self, filter_=None) -> int:
         return self._mongodb_collection.estimated_document_count()
 
     def query(self):
+        _filter = deepcopy(self.filter)
+        # Evaluate `_id` filter field specially, by upcasting to `bson.ObjectId`.
+        if _filter and "_id" in _filter:
+            _filter["_id"] = bson.ObjectId(_filter["_id"])
         data = (
-            self._mongodb_collection.find(filter=self.filter)
+            self._mongodb_collection.find(filter=_filter)
             .batch_size(self.batch_size)
             .skip(self.offset)
             .limit(self.limit)

doc/io/mongodb/loader.md

@@ -109,8 +109,9 @@ appending the HTTP URL query parameter `batch-size` to the source URL, like
 
 ### Filter
 Use the HTTP URL query parameter `filter` on the source URL, like
-`&filter={"exchange": {"$eq": "NASDAQ"}}`, in order to provide a MongoDB
-query filter as a JSON string.
+`&filter={"exchange":{"$eq":"NASDAQ"}}`, or
+`&filter={"_id":"66f0002e98c00fb8261d87c8"}`,
+in order to provide a MongoDB query filter as a JSON string.
 It works in the same way like `mongoexport`'s `--query` option.
 On more complex query expressions, make sure to properly encode the right
 value using URL/Percent Encoding.

tests/io/mongodb/test_copy.py

@@ -3,6 +3,7 @@
 from pathlib import Path
 from unittest import mock
 
+import bson
 import pymongo
 import pytest
 from zyp import CollectionTransformation, MokshaTransformation
@@ -64,7 +65,7 @@ def test_mongodb_copy_server_database(caplog, cratedb, mongodb):
     assert results[0]["data"] == data_out
 
 
-def test_mongodb_copy_server_collection_with_filter(caplog, cratedb, mongodb):
+def test_mongodb_copy_server_collection_with_filter_timestamp(caplog, cratedb, mongodb):
     """
     Verify MongoDB -> CrateDB data transfer for a specific collection, with filtering.
     """
@@ -102,6 +103,54 @@ def test_mongodb_copy_server_collection_with_filter(caplog, cratedb, mongodb):
     assert results[0]["data"] == data_out[1]
 
 
+def test_mongodb_copy_server_collection_with_filter_oid(caplog, cratedb, mongodb):
+    """
+    Verify MongoDB -> CrateDB data transfer for a specific collection, with filtering by oid.
+    """
+
+    # Define source and target URLs.
+    filter_expression = json.dumps({"_id": "66f0002e98c00fb8261d87c8"})
+    mongodb_url = f"{mongodb.get_connection_url()}/testdrive/demo?filter={filter_expression}"
+    cratedb_url = f"{cratedb.get_connection_url()}/testdrive/demo"
+
+    # Define data.
+    data_in = [
+        {
+            "_id": bson.ObjectId("66efff8de45f4f7b695134a6"),
+            "device": "Räuber",
+            "temperature": 42.42,
+            "timestamp": 1563051934000,
+        },
+        {
+            "_id": bson.ObjectId("66f0002e98c00fb8261d87c8"),
+            "device": "Hotzenplotz",
+            "temperature": 84.84,
+            "timestamp": 1563051934100,
+        },
+    ]
+    data_out = deepcopy(data_in)
+    data_out[0].update({"_id": mock.ANY})
+    data_out[1].update({"_id": mock.ANY})
+
+    # Populate source database.
+    client: pymongo.MongoClient = mongodb.get_connection_client()
+    testdrive = client.get_database("testdrive")
+    demo = testdrive.create_collection("demo")
+    demo.insert_many(data_in)
+
+    # Run transfer command.
+    mongodb_copy(
+        mongodb_url,
+        cratedb_url,
+    )
+
+    # Verify data in target database.
+    cratedb.database.refresh_table("testdrive.demo")
+    assert cratedb.database.count_records("testdrive.demo") == 1
+    results = cratedb.database.run_sql("SELECT * FROM testdrive.demo;", records=True)
+    assert results[0]["data"] == data_out[1]
+
+
 def test_mongodb_copy_filesystem_folder_absolute(caplog, cratedb, mongodb):
     """
     Verify MongoDB -> CrateDB data transfer for all files in a folder, with relative addressing.