Skip to content

symfonic.core.observability.mongo_metrics_store

mongo_metrics_store

MongoMetricsStore — durable :class:MetricsStore on the llm_calls collection (v8.8.0).

The MongoDB counterpart to :class:PostgresMetricsStore. Uses motor (async, matching the async fan-out + reads) behind the existing symfonic-core[mongodb] extra. The scaffold owns collection bootstrap by calling :meth:ensure_indexes on startup (idempotent — Mongo is schemaless, no migration). Retention is native: a TTL index on created_at expires old documents automatically, so :meth:prune is an on-demand fallback.

Every read returns the same snapshot dict shape the collector emits, so the scaffold's build_conversation_detail() + response models are reused unchanged across backends.

MongoMetricsStore 

MongoMetricsStore(
    database_factory: DatabaseFactory,
    *,
    retention_days: int = 90,
)

Durable :class:MetricsStore backed by a Mongo llm_calls collection.

Parameters

database_factory Zero-arg callable returning a motor async database (AsyncIOMotorDatabase). Kept as a factory (not a bound db) to match PostgresMetricsStore's framework-agnostic wiring. retention_days TTL for the created_at index created by :meth:ensure_indexes.

Source code in src/symfonic/core/observability/mongo_metrics_store.py 
43
44
45
def __init__(self, database_factory: DatabaseFactory, *, retention_days: int = 90) -> None:
    self._database_factory = database_factory
    self._retention_days = max(1, retention_days)

ensure_indexes 

ensure_indexes() -> None

Create the read + TTL indexes (idempotent). Called on startup.

motor index creation is async, but startup wiring is sync (mirrors the Postgres migration step), so this schedules the coro on the running loop when there is one, else runs it to completion.

Source code in src/symfonic/core/observability/mongo_metrics_store.py 
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
def ensure_indexes(self) -> None:
    """Create the read + TTL indexes (idempotent). Called on startup.

    ``motor`` index creation is async, but startup wiring is sync
    (mirrors the Postgres migration step), so this schedules the coro
    on the running loop when there is one, else runs it to completion.
    """
    import asyncio  # noqa: PLC0415

    async def _create() -> None:
        coll = self._collection()
        await coll.create_index([("tenant_id", 1), ("created_at", -1)])
        await coll.create_index(
            [("tenant_id", 1), ("conversation_id", 1), ("created_at", 1)],
        )
        await coll.create_index(
            [("created_at", 1)],
            expireAfterSeconds=self._retention_days * 86_400,
            name="ttl_created_at",
        )

    try:
        loop = asyncio.get_running_loop()
    except RuntimeError:
        asyncio.run(_create())
    else:
        loop.create_task(_create())

prune async 

prune(cutoff: datetime) -> int

On-demand purge (the TTL index handles routine expiry).

Source code in src/symfonic/core/observability/mongo_metrics_store.py 
88
89
90
91
async def prune(self, cutoff: datetime) -> int:
    """On-demand purge (the TTL index handles routine expiry)."""
    result = await self._collection().delete_many({"created_at": {"$lt": cutoff}})
    return int(getattr(result, "deleted_count", 0) or 0)