MongoDB/PyMongo: Amalgamate PyMongo driver to use CrateDB as backend

.github/workflows/influxdb.yml

@@ -68,7 +68,7 @@ jobs:
         cache: 'pip'
         cache-dependency-path: 'pyproject.toml'
 
-    - name: Setup project
+    - name: Set up project
       run: |
 
         # `setuptools 0.64.0` adds support for editable install hooks (PEP 660).

.github/workflows/main.yml

@@ -20,7 +20,8 @@ concurrency:
 
 jobs:
 
-  tests:
+
+  tests-main:
 
     runs-on: ${{ matrix.os }}
     strategy:
@@ -43,7 +44,10 @@ jobs:
           - 4200:4200
           - 5432:5432
 
-    name: Python ${{ matrix.python-version }} on OS ${{ matrix.os }}
+    name: "
+    Generic:
+    Python ${{ matrix.python-version }} on OS ${{ matrix.os }}
+    "
     steps:
 
     - name: Acquire sources
@@ -57,7 +61,7 @@ jobs:
         cache: 'pip'
         cache-dependency-path: 'pyproject.toml'
 
-    - name: Setup project
+    - name: Set up project
       run: |
 
         # `setuptools 0.64.0` adds support for editable install hooks (PEP 660).
@@ -84,3 +88,68 @@ jobs:
         env_vars: OS,PYTHON
         name: codecov-umbrella
         fail_ci_if_error: true
+
+
+  tests-pymongo:
+
+    runs-on: ${{ matrix.os }}
+    strategy:
+      fail-fast: false
+      matrix:
+        os: ["ubuntu-latest"]
+        python-version: ["3.9", "3.12"]
+
+    env:
+      OS: ${{ matrix.os }}
+      PYTHON: ${{ matrix.python-version }}
+      # Do not tear down Testcontainers
+      TC_KEEPALIVE: true
+
+    # https://docs.github.com/en/actions/using-containerized-services/about-service-containers
+    services:
+      cratedb:
+        image: crate/crate:nightly
+        ports:
+          - 4200:4200
+          - 5432:5432
+
+    name: "
+    PyMongo:
+    Python ${{ matrix.python-version }} on OS ${{ matrix.os }}"
+    steps:
+
+    - name: Acquire sources
+      uses: actions/checkout@v4
+
+    - name: Set up Python
+      uses: actions/setup-python@v4
+      with:
+        python-version: ${{ matrix.python-version }}
+        architecture: x64
+        cache: 'pip'
+        cache-dependency-path: 'pyproject.toml'
+
+    - name: Set up project
+      run: |
+
+        # `setuptools 0.64.0` adds support for editable install hooks (PEP 660).
+        # https://github.com/pypa/setuptools/blob/main/CHANGES.rst#v6400
+        pip install "setuptools>=64" --upgrade
+
+        # Install package in editable mode.
+        pip install --use-pep517 --prefer-binary --editable=.[pymongo,test,develop]
+
+    - name: Run linter and software tests
+      run: |
+        pytest -m pymongo
+
+    - name: Upload coverage to Codecov
+      uses: codecov/codecov-action@v4
+      env:
+        CODECOV_TOKEN: ${{ secrets.CODECOV_TOKEN }}
+      with:
+        files: ./coverage.xml
+        flags: pymongo
+        env_vars: OS,PYTHON
+        name: codecov-umbrella
+        fail_ci_if_error: true

.github/workflows/mongodb.yml

@@ -62,7 +62,7 @@ jobs:
         cache: 'pip'
         cache-dependency-path: 'pyproject.toml'
 
-    - name: Setup project
+    - name: Set up project
       run: |
 
         # `setuptools 0.64.0` adds support for editable install hooks (PEP 660).

.gitignore

@@ -2,6 +2,7 @@
 .eggs
 .env
 .idea
+.mypy_cache
 *.pyc
 .venv*
 __pycache__

CHANGES.md

@@ -4,6 +4,7 @@
 ## Unreleased
 - IO: Added the `if-exists` query parameter by updating to influxio 0.4.0.
 - Rockset: Added CrateDB Rockset Adapter, a HTTP API emulation layer
+- MongoDB: Added adapter amalgamating PyMongo to use CrateDB as backend
 
 ## 2024/06/18 v0.0.14
 - Add `ctk cfr` and `ctk wtf` diagnostics programs

cratedb_toolkit/adapter/pymongo/__init__.py

@@ -0,0 +1 @@
+from .api import PyMongoCrateDBAdapter  # noqa: F401

cratedb_toolkit/adapter/pymongo/api.py

@@ -0,0 +1,77 @@
+from unittest.mock import patch
+
+import pymongo.collection
+
+from cratedb_toolkit.adapter.pymongo.collection import collection_factory
+from cratedb_toolkit.sqlalchemy.patch import patch_types_map
+from cratedb_toolkit.util import DatabaseAdapter
+from cratedb_toolkit.util.pandas import patch_pandas_sqltable_with_extended_mapping
+
+
+class PyMongoCrateDBAdapter:
+    """
+    Patch PyMongo to talk to CrateDB.
+    """
+
+    def __init__(self, dburi: str):
+        self.cratedb = DatabaseAdapter(dburi=dburi)
+        self.collection_backup = pymongo.collection.Collection
+
+        collection_patched = collection_factory(cratedb=self.cratedb)  # type: ignore[misc]
+        self.patches = [
+            # Patch PyMongo's `Collection` implementation.
+            patch("pymongo.collection.Collection", collection_patched),
+            patch("pymongo.database.Collection", collection_patched),
+            # Converge a few low-level functions of PyMongo to no-ops.
+            patch("pymongo.mongo_client.MongoClient._ensure_session"),
+            patch("pymongo.mongo_client._ClientConnectionRetryable._get_server"),
+        ]
+
+    def start(self):
+        self.adjust_sqlalchemy()
+        self.activate()
+
+    def stop(self):
+        self.deactivate()
+
+    def __enter__(self):
+        self.start()
+
+    def __exit__(self, exc_type, exc_val, exc_tb):
+        self.stop()
+
+    def adjust_sqlalchemy(self):
+        """
+        Configure CrateDB SQLAlchemy dialect.
+
+        Setting the CrateDB column policy to `dynamic` means that new columns
+        can be added without needing to explicitly change the table definition
+        by running corresponding `ALTER TABLE` statements.
+
+        https://cratedb.com/docs/crate/reference/en/latest/general/ddl/column-policy.html#dynamic
+        """
+        # 1. Patch data types for CrateDB dialect.
+        # TODO: Upstream to `sqlalchemy-cratedb`.
+        patch_types_map()
+
+        # 2. Prepare pandas.
+        # TODO: Provide unpatching hook.
+        # TODO: Use `with table_kwargs(...)`.
+        from cratedb_toolkit.util.pandas import patch_pandas_sqltable_with_dialect_parameters
+
+        patch_pandas_sqltable_with_dialect_parameters(table_kwargs={"crate_column_policy": "'dynamic'"})
+        patch_pandas_sqltable_with_extended_mapping()
+
+    def activate(self):
+        """
+        Swap in the MongoDB -> CrateDB adapter, by patching functions in PyMongo.
+        """
+        for patch_ in self.patches:
+            patch_.start()
+
+    def deactivate(self):
+        """
+        Swap out the MongoDB -> CrateDB adapter, by restoring patched functions.
+        """
+        for patch_ in self.patches:
+            patch_.stop()

cratedb_toolkit/adapter/pymongo/backlog.md

@@ -0,0 +1,59 @@
+# PyMongo CrateDB Adapter Backlog
+
+## Iteration +1
+- Upstream / converge patches.
+  - `cratedb_toolkit/sqlalchemy/patch.py`
+  - `cratedb_toolkit/util/pandas.py`
+  - `cratedb_toolkit/adapter/pymongo/api.py::adjust_sqlalchemy`
+  - `cratedb_toolkit/adapter/pymongo/collection.py::insert_returning_id`
+- Make `jessiql` work with more recent SQLAlchemy 2.x.
+- Make adapter work with pandas 2.2.
+
+## Iteration +2
+- Add documentation.
+- Add missing essential querying features: Examples: sort order, skip, limit
+- Add missing essential methods. Example: `db.my_collection.drop()`.
+
+## Iteration +2
+- Make write-synchronization behavior (refresh table) configurable.
+- Handle deeply nested documents of various types.
+- Unlock more features from canonical examples.
+  https://pymongo.readthedocs.io/en/stable/examples/
+
+## Iteration +3
+- Decoding timestamps are yielding only Decimals instead of datetime objects
+  when printed on the terminal? See example program.
+  There is also a warning:
+  ```shell
+  SAWarning: Dialect crate+crate-python does *not* support Decimal objects natively,
+  and SQLAlchemy must convert from floating point - rounding errors and other issues
+  may occur. Please consider storing Decimal numbers as strings or integers on this
+  platform for lossless storage.
+  ```
+  So, why are `Decimal` types used here, at all?
+- Mimic MongoDB exceptions.
+  Example:
+  ```python
+  jessiql.exc.InvalidColumnError: Invalid column "x" for "Surrogate" specified in sort
+  ```
+
+
+## Done
+
+- Make it work
+- Using individual columns for fields does not work, because `insert_one` works
+  iteratively, and needs to evolve the table schema gradually. As a consequence,
+  we need to use `OBJECT(DYNAMIC)` for storing MongoDB fields.
+- Add software tests
+
+### Research
+How to translate a MongoDB query expression?
+
+- https://github.com/gordonbusyman/mongo-to-sql-converter
+- https://github.com/2do2go/json-sql
+
+- https://github.com/kolypto/py-mongosql
+- https://github.com/SY-Xuan/mongo2sql
+- https://github.com/nsragow/MongoToSqlParse
+- https://github.com/sushmitharao2124/MongoToSQLConverter
+- https://github.com/Phomint/MongoSQL