Update pymongo3_durable_cursor.py

.gitignore

@@ -21,6 +21,12 @@ var/
 *.egg
 .eggs/
 
+# Virtualenv
+.venv/
+
+# Pytest cache
+.pytest_cache/
+
 # Installer logs
 pip-log.txt
 pip-delete-this-directory.txt

MANIFEST.in

@@ -1 +1,3 @@
 include *.txt *.md *.rst
+include LICENSE
+include NOTICE

NOTICE

@@ -0,0 +1,14 @@
+MongoDBProxy
+Copyright 2013 Gustav Arngarden
+
+This product includes software developed by
+Gustav Arngarden (https://github.com/arngarden).
+
+==========================================================================
+
+This work is a derivative of the original MongoDBProxy.
+Modifications to support modern Python 3 and PyMongo 4 versions,
+along with a comprehensive test suite and other enhancements,
+were made by Martin Alge (https://github.com/Alge).
+
+Copyright 2025 Martin Alge

README.rst

@@ -1,30 +1,36 @@
 MongoDBProxy
 ============
 
-MongoDBProxy is used to create a proxy around a MongoDB-connection in order to
-automatically handle AutoReconnect-exceptions.  You use MongoDBProxy in the
-same way you would an ordinary MongoDB-connection but don't need to worry about
+MongoDBProxy is used to create a proxy around a MongoDB connection in order to
+automatically handle AutoReconnect exceptions. You use MongoDBProxy in the
+same way you would an ordinary MongoDB connection but don't need to worry about
 handling AutoReconnects by yourself.
 
 Usage::
 
     >>> import pymongo
-    >>> import mongo_proxy
-    >>> safe_conn = mongo_proxy.MongoProxy(pymongo.MongoReplicaSetClient(replicaSet='blog_rs')
-    >>> safe_conn.blogs.posts.insert(post)  # Automatically handles AutoReconnect.
+    >>> from mongo_proxy import MongoProxy
+    >>>
+    >>> client = pymongo.MongoClient(replicaSet='blog_rs')
+    >>> safe_conn = MongoProxy(client)
+    >>> safe_conn.blogs.posts.insert_one({'some': 'post'})  # Automatically handles AutoReconnect.
 
+Fork Information
+----------------
 
-**See here for more details:**
-`<http://www.arngarden.com/2013/04/29/handling-mongodb-autoreconnect-exceptions-in-python-using-a-proxy/>`_
-
-**Contributors**:
-
-- Jonathan Kamens (`<https://github.com/jikamens>`_)
-- Michael Cetrulo (`<https://github.com/git2samus>`_)
-- Richard Frank (`<https://github.com/richafrank>`_)
-- David Lindquist (`<https://github.com/dlindquist>`_)
+This is a modernized fork of Gustav Arngarden's original MongoDBProxy. The primary goal of this version is to provide a stable, well-tested proxy compatible with modern Python 3 environments while maintaining support for legacy MongoDB databases.
 
 Installation
 ------------
 
-pip3 install MongoDBProxy-official
+To install the package with its testing dependencies, run:
+
+    pip install -e .[test]
+
+
+Compatibility
+--------------
+
+This library is compatible with **Python 3.6+** and **PyMongo 3.12+** (version < 4.0).
+
+This focus on PyMongo 3.x is a deliberate choice to ensure compatibility with older MongoDB server versions, such as MongoDB 3.4, which are not supported by PyMongo 4.x.

example.py

@@ -0,0 +1,87 @@
+import logging
+import time
+from pymongo import MongoClient
+from mongo_proxy import MongoProxy
+
+
+def main():
+    # --- Configuration ---
+    # Define your MongoDB replica set nodes here
+    mongo_nodes = [
+        "192.168.0.113:27017",
+        "192.168.0.113:27018",
+        "192.168.0.113:27019",
+    ]
+    # Define your replica set name
+    replica_set_name = "rs0"
+    # --- End Configuration ---
+
+    # Configure basic logging to capture WARNING and above
+    logging.basicConfig(
+        level=logging.WARNING,
+        format='%(asctime)s - %(name)s - %(levelname)s - %(message)s'
+    )
+
+    # Enable DEBUG logging specifically for the mongo_proxy module
+    proxy_logger = logging.getLogger("mongo_proxy")
+    proxy_logger.setLevel(logging.DEBUG)
+
+    # Also enable INFO level for our script
+    script_logger = logging.getLogger(__name__)
+    script_logger.setLevel(logging.INFO)
+
+    # Add a console handler to our script logger to see our messages
+    if not script_logger.handlers:
+        handler = logging.StreamHandler()
+        handler.setLevel(logging.INFO)
+        formatter = logging.Formatter(
+            '%(asctime)s - %(name)s - %(levelname)s - %(message)s')
+        handler.setFormatter(formatter)
+        script_logger.addHandler(handler)
+        script_logger.propagate = False
+
+    # Connect to the MongoDB replica set using the list of nodes
+    script_logger.info(
+        f"Attempting to connect to replica set '{replica_set_name}' with nodes: {mongo_nodes}")
+    client = MongoClient(mongo_nodes, replicaSet=replica_set_name)
+
+    # Test the connection
+    try:
+        # The ismaster command is cheap and does not require auth.
+        client.admin.command('ismaster')
+        script_logger.info("✓ Successfully connected to MongoDB replica set")
+    except Exception as e:
+        script_logger.error(f"✗ Failed to connect: {e}")
+        return
+
+    # Create MongoProxy - this handles AutoReconnect automatically
+    proxy = MongoProxy(client)
+
+    collection = proxy.testdb.mycollection
+
+    script_logger.info("MongoProxy example started.")
+    script_logger.info(
+        "Try: rs.stepDown() in the mongo shell to trigger a failover and see reconnection logs.")
+    script_logger.info("Press Ctrl+C to exit.\n")
+
+    counter = 0
+    while True:
+        try:
+            counter += 1
+
+            # Simple database operation
+            doc = {"counter": counter, "timestamp": time.time()}
+            result = collection.insert_one(doc)
+            script_logger.info(f"[{counter}] Inserted document: {result.inserted_id}")
+
+        except KeyboardInterrupt:
+            script_logger.info("\nExiting...")
+            break
+        except Exception as e:
+            script_logger.error(f"[{counter}] An error occurred: {e}")
+
+        time.sleep(3)
+
+
+if __name__ == "__main__":
+    main()

manual_tests.py

@@ -0,0 +1,147 @@
+import logging
+import time
+import random
+from pymongo import MongoClient, errors
+from mongo_proxy import MongoProxy, DurableCursor
+
+# --- ‼️ IMPORTANT CONFIGURATION ‼️ ---
+# Change this to the connection string for your replica set.
+MONGO_URI = "mongodb://192.168.0.116:27017,192.168.0.116:27018,192.168.0.116:27019/"
+REPLICA_SET = "rs0" # Change this to your replica set name.
+# --- End Configuration ---
+
+
+def setup_logging():
+    """Configures logging to show detailed info from the proxy."""
+    logging.basicConfig(
+        level=logging.INFO,
+        format='%(asctime)s - %(name)s - %(levelname)s - %(message)s'
+    )
+    logging.getLogger("mongo_proxy").setLevel(logging.DEBUG)
+    # Silence the noisy pymongo connection pool logs for this test
+    logging.getLogger("pymongo.pool").setLevel(logging.WARNING)
+
+
+def test_writer_resilience():
+    # ... (this function remains the same and is correct)
+    print("\n--- 🚀 Starting Write Resilience Test ---")
+    print("This test will insert a new document every 4 seconds.")
+    print("While it's running, perform actions on your MongoDB cluster, such as:")
+    print("  - Use `rs.stepDown()` in mongosh to trigger a failover.")
+    print("  - Stop the primary node's container (`docker stop ...`).")
+    print("Watch the output to see the proxy handle the errors and reconnect.\n")
+
+    client = MongoClient(MONGO_URI, replicaSet=REPLICA_SET, serverSelectionTimeoutMS=5000)
+    proxy = MongoProxy(client)
+    collection = proxy.testdb.manual_test_collection
+
+    try:
+        collection.drop()
+        print("✓ Dropped old test collection.")
+    except Exception as e:
+        print(f"✗ Could not drop collection (this is okay on first run): {e}")
+
+    counter = 0
+    while True:
+        try:
+            counter += 1
+            doc = {"counter": counter, "time": time.ctime()}
+            result = collection.insert_one(doc)
+            print(f"[{counter}] ✓ Inserted document: {result.inserted_id}")
+
+        except KeyboardInterrupt:
+            print("\n--- 🛑 Test stopped by user. ---")
+            break
+        except Exception as e:
+            print(f"[{counter}] ✗ An error occurred: {type(e).__name__} - {e}")
+            print(f"[{counter}] ⏳ Proxy will now attempt to reconnect...")
+
+        time.sleep(4)
+
+
+def test_durable_cursor():
+    """
+    Tests the DurableCursor's ability to survive a failover mid-iteration
+    without losing its place.
+    """
+    print("\n--- 🚀 Starting Durable Cursor Test ---")
+
+    client = MongoClient(MONGO_URI, replicaSet=REPLICA_SET)
+    setup_collection = client.testdb.durable_cursor_test
+
+    # ** THE FIX: Use a number of documents GREATER than the default batch size (101). **
+    num_docs = 300
+    failover_point = 105 # A point safely after the first batch is exhausted.
+
+    print(f"Setting up the collection with {num_docs} documents...")
+    setup_collection.drop()
+    setup_collection.insert_many([{'doc_num': i} for i in range(1, num_docs + 1)])
+    print("✓ Collection setup complete.")
+
+    print(f"\nStarting slow iteration (1 doc every 2 seconds).")
+    print("\n‼️ WATCH THE COUNTER ‼️")
+    print(f"TRIGGER A FAILOVER AFTER YOU SEE DOCUMENT #{failover_point} IS RETRIEVED.")
+    print("This guarantees the next fetch will require a network call.\n")
+
+    proxy = MongoProxy(client)
+    proxied_collection = proxy.testdb.durable_cursor_test
+
+    retrieved_docs = []
+    try:
+        # We still use batch_size=1 to be explicit, but the number of documents is the key.
+        durable_cursor = DurableCursor(
+            proxied_collection,
+            sort=[('doc_num', 1)],
+            batch_size=1 
+        )
+
+        for doc in durable_cursor:
+            print(f"[Cursor] -> Retrieved document {doc['doc_num']}/{num_docs}", end='\r')
+            retrieved_docs.append(doc['doc_num'])
+
+            if doc['doc_num'] == failover_point:
+                print(f"\n[!] NOW IS A GOOD TIME TO TRIGGER THE FAILOVER (`rs.stepDown()`) [!]")
+
+            time.sleep(2)
+
+        print(f"\n\n--- ✅ Test Complete ---")
+        if len(retrieved_docs) == num_docs and sorted(retrieved_docs) == list(range(1, num_docs + 1)):
+            print(f"🎉 SUCCESS! All {num_docs} documents were retrieved in order without duplicates.")
+        else:
+            print(f"🔥 FAILURE! Expected {num_docs} unique documents, but got {len(retrieved_docs)}.")
+            print(f"   The first few retrieved documents: {sorted(retrieved_docs)[:20]}...")
+
+    except KeyboardInterrupt:
+        print("\n--- 🛑 Test stopped by user. ---")
+    except Exception as e:
+        print(f"\n--- ✗ TEST FAILED WITH AN UNEXPECTED ERROR ---")
+        print(f"Error: {type(e).__name__} - {e}")
+        import traceback
+        traceback.print_exc()
+
+
+def main():
+    setup_logging()
+    print("========================================")
+    print("  MongoDBProxy Manual Test Suite")
+    print("========================================")
+
+    while True:
+        print("\nChoose a test to run:")
+        print("  [1] Write Resilience Test (Failover/Outage)")
+        print("  [2] Durable Cursor Resilience Test")
+        print("  [q] Quit")
+
+        choice = input("> ")
+
+        if choice == '1':
+            test_writer_resilience()
+        elif choice == '2':
+            test_durable_cursor()
+        elif choice.lower() == 'q':
+            break
+        else:
+            print("Invalid choice, please try again.")
+
+if __name__ == "__main__":
+    main()

mongo_proxy/__init__.py

@@ -1,10 +1,8 @@
 from .mongodb_proxy import MongoProxy
 from .durable_cursor import DurableCursor, MongoReconnectFailure
-from .pymongo3_durable_cursor import PyMongo3DurableCursor
 
 __all__ = [
     'MongoProxy',
     'DurableCursor',
     'MongoReconnectFailure',
-    'PyMongo3DurableCursor',
-]
+]