openframe-adapters-db-mongo

MongoDB document store adapter for the OpenFrame Microservice Suite.

Part of the openframe-adapters monorepo. Implements BaseRepository[T] and HealthCheck from openframe-core using motor (AsyncIOMotorClient).

Installation

pip install openframe-adapters-db-mongo

Required env vars:

MONGO_URL=mongodb://user:password@host:27017
MONGO_DATABASE=mydb

Atlas SRV URIs work too:

MONGO_URL=mongodb+srv://user:password@cluster.mongodb.net

Quick start

Raw dict mode

from openframe.adapters.db.mongo import MongoSettings, MongoRepository

settings = MongoSettings()  # reads MONGO_URL and MONGO_DATABASE from env
repo = MongoRepository(settings, collection="artifacts")

doc = await repo.get("507f1f77bcf86cd799439011")   # dict | None
docs, total = await repo.list(10, 0)                # ([dict, ...], int)
created = await repo.create({"title": "paper"})
updated = await repo.update({"_id": "...", "title": "updated"})
deleted = await repo.delete("507f1f77bcf86cd799439011")  # bool

Typed domain mode

from dataclasses import dataclass
from openframe.adapters.db.mongo import MongoSettings, MongoRepository

@dataclass
class Artifact:
    id: str
    title: str
    artifact_type: str

class ArtifactRepository(MongoRepository[Artifact]):
    _collection = "artifacts"

    def _doc_to_entity(self, doc: dict) -> Artifact:
        return Artifact(
            id=doc["_id"],
            title=doc["title"],
            artifact_type=doc["artifact_type"],
        )

    def _entity_to_doc(self, entity: Artifact) -> dict:
        return {
            "_id": entity.id,
            "title": entity.title,
            "artifact_type": entity.artifact_type,
        }

settings = MongoSettings()
repo = ArtifactRepository(settings)
artifact: Artifact | None = await repo.get("507f1f77bcf86cd799439011")

Wiring into an application

The examples above construct MongoRepository directly and skip lifecycle management. For a real service, wire the adapter through its MongoPlugin (the BasePort-satisfying plugin class) so connect/health/close are handled for you. The recommended entry point is ApplicationBootstrap.compose() from openframe-core>=3.3:

from openframe.core.runtime import ApplicationBootstrap
from openframe.core.ports import Capability
from openframe.adapters.db.mongo import MongoPlugin, MongoSettings

mongo_port = MongoPlugin(MongoSettings(), collection="artifacts")

async with ApplicationBootstrap.compose(mongo_port) as app:
    repo = app.get(Capability.PERSISTENCE).get_repository()
    doc = await repo.get("507f1f77bcf86cd799439011")
# initialize() ran on entry, shutdown() runs automatically on exit

compose() is the right choice for the common case of one or a few ports. Reach for a subclassed ApplicationBootstrap with configure() only when you need per-port config=/init_timeout= or conditional registration order, and use ApplicationBootstrap.registry as an escape hatch for anything neither tier covers (e.g. registry.get_all(Capability.PERSISTENCE) when more than one persistence port is registered deliberately).

_id handling

MongoDB uses _id as the document identifier; BaseRepository uses entity_id: str. The adapter bridges this transparently:

• Reading: _id is always returned as str — ObjectId is never exposed to callers. A convenience id key is also added mirroring _id.
• Writing: If an entity dict has an id key but no _id key, the adapter maps id → _id before insertion.
• Filtering: String entity IDs that are valid 24-char hex are parsed as ObjectId for indexed lookups; other strings are used as plain string _id values.

Configuration

All settings are read from environment variables.

Timeout strategy

Two layers of timeout on every operation:

1. asyncio.timeout(settings.operation_timeout) — cancels the Python coroutine.
2. max_time_ms=int(settings.operation_timeout * 1000) passed to motor query methods — instructs the MongoDB server to abort the query.

Health checks

MongoRepository implements the HealthCheck protocol from openframe-core.

alive = await repo.ping()       # admin.command("ping") — fast liveness check
ready = await repo.is_ready()   # list_collection_names() — full readiness check

Both methods return False on any failure and never raise.

Exception hierarchy

All exceptions are AdapterError subclasses from openframe.core.exceptions. Raw motor/pymongo exceptions never escape the adapter.

Development

# from the package directory
pip install -e ".[dev]"
pytest tests/ -v

Protocol conformance

from openframe.core.ports import BaseRepository
from openframe.core.health import HealthCheck

repo = MongoRepository(settings, collection="artifacts")
assert isinstance(repo, BaseRepository)   # True — structural check
assert isinstance(repo, HealthCheck)      # True — structural check

No inheritance from either Protocol is required or used.

License

MIT