Skip to content

Commit 24e95fa

Browse files
rkingsburyclaude
andcommitted
Back MemoryStore with a real MongoDB when available, fall back to mongomock-ng
MemoryStore now probes host:port (default localhost:27017) on connect and, if a MongoDB server answers, stores data in a real, ephemeral MongoDB database for full MongoDB compatibility and performance. Each instance uses a unique maggma_memory_<uuid> database that is dropped automatically (via weakref finalizer) when the store is garbage collected or the interpreter exits, so the user never has to set up or clean up a database. If no server is reachable the store transparently falls back to an in-process mongomock-ng database. - close() is now a no-op that leaves the store usable, preserving the historical mongomock behavior relied upon by the builders (pymongo raises on use-after-close). Use connect(force_reset=True) to discard contents. - JSONStore/FileStore inherit the behavior via a shared _connect_collection; MontyStore gets its own close() so its disk-backed client still closes. - Adds backward-compatible host/port/mongoclient_kwargs/server_selection_timeout_ms kwargs; server_selection_timeout_ms=0 forces the mongomock backend. Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>
1 parent ca564e4 commit 24e95fa

2 files changed

Lines changed: 173 additions & 10 deletions

File tree

src/maggma/stores/mongolike.py

Lines changed: 126 additions & 7 deletions
Original file line numberDiff line numberDiff line change
@@ -5,10 +5,12 @@
55
"""
66

77
import warnings
8+
import weakref
89
from collections.abc import Callable, Iterator
910
from itertools import chain, groupby
1011
from pathlib import Path
1112
from typing import Any, Literal
13+
from uuid import uuid4
1214

1315
import bson
1416
import mongomock_ng as mongomock
@@ -505,23 +507,126 @@ def connect(self, force_reset: bool = False):
505507

506508
class MemoryStore(MongoStore):
507509
"""
508-
An in-memory Store that functions similarly
509-
to a MongoStore.
510+
An in-memory Store that functions similarly to a MongoStore.
511+
512+
If a MongoDB server is reachable (by default on ``localhost:27017``), the
513+
data is stored in a real, ephemeral MongoDB database for full MongoDB
514+
compatibility and performance. That database is namespaced uniquely per
515+
Store instance and is dropped automatically when the Store is garbage
516+
collected or the interpreter exits, so the user never has to create or
517+
clean up a database manually. If no server is reachable, the Store
518+
transparently falls back to an in-process ``mongomock`` database, so it
519+
works even with no MongoDB installed.
510520
"""
511521

512-
def __init__(self, collection_name: str = "memory_db", **kwargs):
522+
#: Set to True by connect() when a real MongoDB backend is in use. Defined
523+
#: at class level so close() is safe on subclasses (e.g. MontyStore) that
524+
#: do not call MemoryStore.__init__.
525+
_using_real_mongo: bool = False
526+
_finalizer = None
527+
528+
def __init__(
529+
self,
530+
collection_name: str = "memory_db",
531+
host: str = "localhost",
532+
port: int = 27017,
533+
mongoclient_kwargs: dict | None = None,
534+
server_selection_timeout_ms: int = 500,
535+
**kwargs,
536+
):
513537
"""
514538
Initializes the Memory Store.
515539
516540
Args:
517541
collection_name: name for the collection in memory.
542+
host: hostname to probe for a running MongoDB server to back the
543+
Store with. The data is written to an ephemeral database that
544+
is dropped when the Store is closed.
545+
port: TCP port to probe for a running MongoDB server.
546+
mongoclient_kwargs: Dict of extra kwargs to pass to MongoClient when
547+
a real MongoDB backend is used.
548+
server_selection_timeout_ms: how long, in milliseconds, to wait when
549+
probing ``host:port`` for a MongoDB server before falling back
550+
to an in-process ``mongomock`` database. Set to 0 (or less) to
551+
skip the probe entirely and always use ``mongomock``.
518552
"""
519553
self.collection_name = collection_name
554+
self.host = host
555+
self.port = port
556+
self.mongoclient_kwargs = mongoclient_kwargs or {}
557+
self.server_selection_timeout_ms = server_selection_timeout_ms
558+
# unique, ephemeral database name so that multiple MemoryStore instances
559+
# backed by the same real MongoDB server do not clobber one another
560+
self._database = f"maggma_memory_{uuid4().hex}"
520561
self.default_sort = None
521562
self._coll = None
522563
self.kwargs = kwargs
523564
super(MongoStore, self).__init__(**kwargs)
524565

566+
def _get_memory_client(self) -> MongoClient:
567+
"""
568+
Return a client backing the in-memory Store.
569+
570+
Attempts to connect to a real MongoDB server at ``host:port`` for full
571+
MongoDB compatibility and performance. If none is reachable within
572+
``server_selection_timeout_ms``, falls back to an in-process
573+
``mongomock`` client so the Store works without a MongoDB server.
574+
"""
575+
if self.server_selection_timeout_ms > 0:
576+
mongoclient_kwargs = dict(self.mongoclient_kwargs)
577+
mongoclient_kwargs.setdefault("serverSelectionTimeoutMS", self.server_selection_timeout_ms)
578+
try:
579+
client = MongoClient(host=self.host, port=self.port, **mongoclient_kwargs)
580+
# force server selection to confirm a server is actually reachable
581+
client.admin.command("ping")
582+
self._using_real_mongo = True
583+
# ensure the ephemeral database is dropped when this Store is
584+
# garbage collected or the interpreter exits, so nothing is left
585+
# behind on the server
586+
if self._finalizer is None:
587+
self._finalizer = weakref.finalize(
588+
self,
589+
self._drop_ephemeral_database,
590+
self.host,
591+
self.port,
592+
self._database,
593+
dict(mongoclient_kwargs),
594+
)
595+
self.logger.debug(f"{self.name} using real MongoDB backend at {self.host}:{self.port}")
596+
return client
597+
except Exception:
598+
self.logger.debug(f"{self.name}: no MongoDB server reachable, falling back to mongomock")
599+
600+
self._using_real_mongo = False
601+
return mongomock.MongoClient() # type: ignore
602+
603+
@staticmethod
604+
def _drop_ephemeral_database(host: str, port: int, database: str, mongoclient_kwargs: dict):
605+
"""Drop an ephemeral MongoDB database. Used as a weakref finalizer, so it
606+
must not hold a reference to the Store."""
607+
mongoclient_kwargs.setdefault("serverSelectionTimeoutMS", 500)
608+
try:
609+
client = MongoClient(host=host, port=port, **mongoclient_kwargs)
610+
client.drop_database(database)
611+
client.close()
612+
except Exception:
613+
pass
614+
615+
def _connect_collection(self, force_reset: bool = False):
616+
"""Establish (or re-establish) the underlying in-memory collection."""
617+
if force_reset and self._coll is not None:
618+
old_client = self._coll.database.client
619+
# on a forced reset, discard the previous contents (matching the
620+
# historical mongomock behavior of starting from a fresh client)
621+
if getattr(self, "_using_real_mongo", False):
622+
try:
623+
old_client.drop_database(self._database)
624+
except Exception:
625+
pass
626+
old_client.close()
627+
client = self._get_memory_client()
628+
self._coll = client[self._database][self.collection_name] # type: ignore
629+
525630
def connect(self, force_reset: bool = False):
526631
"""
527632
Connect to the source data.
@@ -531,11 +636,20 @@ def connect(self, force_reset: bool = False):
531636
already connected.
532637
"""
533638
if self._coll is None or force_reset:
534-
self._coll = mongomock.MongoClient().db[self.name] # type: ignore
639+
self._connect_collection(force_reset=force_reset)
535640

536641
def close(self):
537-
"""Close up all collections."""
538-
self._coll.database.client.close()
642+
"""Close up all collections.
643+
644+
For an in-memory Store this is intentionally a no-op that leaves the
645+
Store usable, matching the historical ``mongomock`` behavior (its
646+
``close()`` did nothing) that callers such as the builders rely on when
647+
they query a Store after it has been closed. Resources are released and
648+
any ephemeral MongoDB database backing the Store is dropped when the
649+
Store is garbage collected or at interpreter exit (see
650+
``_drop_ephemeral_database``). Use ``force_reset=True`` on ``connect()``
651+
to explicitly discard the contents and start fresh.
652+
"""
539653

540654
@property
541655
def name(self):
@@ -682,7 +796,7 @@ def connect(self, force_reset: bool = False):
682796
on systems with slow storage when multiple connect / disconnects are performed.
683797
"""
684798
if self._coll is None or force_reset:
685-
self._coll = mongomock.MongoClient().db[self.name] # type: ignore
799+
self._connect_collection(force_reset=force_reset)
686800

687801
# create the .json file if it does not exist
688802
if not self.read_only and not Path(self.paths[0]).exists():
@@ -877,6 +991,11 @@ def connect(self, force_reset: bool = False):
877991
client = MontyClient(self.database_path, **self.client_kwargs)
878992
self._coll = client[self.database_name][self.collection_name]
879993

994+
def close(self):
995+
"""Close up the MontyDB client."""
996+
if self._coll is not None:
997+
self._coll.database.client.close()
998+
880999
@property
8811000
def name(self) -> str:
8821001
"""Return a string representing this data source."""

tests/stores/test_mongolike.py

Lines changed: 47 additions & 3 deletions
Original file line numberDiff line numberDiff line change
@@ -248,13 +248,53 @@ def test_mongostore_newer_in(mongostore):
248248

249249

250250
# Memory store tests
251-
def test_memory_store_connect():
252-
memorystore = MemoryStore()
251+
def test_memory_store_connect_mongomock_fallback():
252+
# server_selection_timeout_ms=0 skips the probe and forces the mongomock backend
253+
memorystore = MemoryStore(server_selection_timeout_ms=0)
253254
assert memorystore._coll is None
254255
memorystore.connect()
256+
assert memorystore._using_real_mongo is False
257+
assert isinstance(memorystore._collection, mongomock_ng.collection.Collection)
258+
259+
260+
def test_memory_store_connect_unreachable_falls_back():
261+
# an unreachable server should fall back to mongomock rather than raise
262+
memorystore = MemoryStore(port=1, server_selection_timeout_ms=50)
263+
memorystore.connect()
264+
assert memorystore._using_real_mongo is False
255265
assert isinstance(memorystore._collection, mongomock_ng.collection.Collection)
256266

257267

268+
def test_memory_store_uses_real_mongo_when_available():
269+
# a real MongoDB server is available in CI; when present it should be used
270+
try:
271+
pymongo.MongoClient(serverSelectionTimeoutMS=500).admin.command("ping")
272+
except Exception:
273+
pytest.skip("no MongoDB server reachable on localhost:27017")
274+
275+
memorystore = MemoryStore()
276+
memorystore.connect()
277+
assert memorystore._using_real_mongo is True
278+
assert isinstance(memorystore._collection, pymongo.collection.Collection)
279+
280+
# the ephemeral database exists while connected
281+
verify_client = pymongo.MongoClient(serverSelectionTimeoutMS=500)
282+
memorystore.update({"task_id": 1, "val": 2})
283+
assert memorystore._database in verify_client.list_database_names()
284+
285+
# data survives close() (the Store remains usable), matching the historical
286+
# in-memory behavior relied upon by builders
287+
memorystore.close()
288+
memorystore.connect()
289+
assert memorystore.count() == 1
290+
291+
# the ephemeral database is dropped when the Store is finalized
292+
database_name = memorystore._database
293+
memorystore._finalizer()
294+
assert database_name not in verify_client.list_database_names()
295+
verify_client.close()
296+
297+
258298
def test_groupby(memorystore):
259299
memorystore.update(
260300
[
@@ -522,8 +562,11 @@ def test_jsonstore_orjson_options(test_dir):
522562
class SubFloat(float):
523563
pass
524564

565+
# Force the mongomock backend (server_selection_timeout_ms=0): a real MongoDB
566+
# backend coerces the SubFloat subclass to a plain float on write/read, so the
567+
# serialization_default option this test exercises would never be triggered.
525568
with ScratchDir("."):
526-
jsonstore = JSONStore("d.json", read_only=False)
569+
jsonstore = JSONStore("d.json", read_only=False, server_selection_timeout_ms=0)
527570
jsonstore.connect()
528571
with pytest.raises(orjson.JSONEncodeError):
529572
jsonstore.update({"wrong_field": SubFloat(1.1), "task_id": 3})
@@ -534,6 +577,7 @@ class SubFloat(float):
534577
read_only=False,
535578
serialization_option=None,
536579
serialization_default=lambda x: "test",
580+
server_selection_timeout_ms=0,
537581
)
538582
jsonstore.connect()
539583
jsonstore.update({"wrong_field": SubFloat(1.1), "task_id": 3})

0 commit comments

Comments
 (0)