Skip to content

symfonic.memory.backends.mongo_edge_indexes

mongo_edge_indexes

Explicit deployment of Mongo graph edge indexes.

Call provision_edge_indexes from a migration or startup job. It is never called by graph reads. Schema-write deployments pass schema_writes=True; read-only deployments pass False to verify that provisioning is complete. For tenants whose edge collection does not yet exist, supply tenant_ids. The caller must rerun provisioning when a later tenant is introduced.

provision_edge_indexes async 

provision_edge_indexes(database: Any, *, collection_prefix: str = 'graph', tenant_ids: Iterable[str] = (), schema_writes: bool) -> tuple[str, ...]

Apply or verify the two ordered indexes on each tenant edge collection.

Existing collections are discovered from database. Explicit tenant IDs cover new collections before their first edge write. Repeating apply mode is safe: MongoDB treats a matching index name and key spec as a no-op. Verify mode uses only listCollections and listIndexes privileges; it fails when a requested tenant has no collection or required index.

Source code in src/symfonic/memory/backends/mongo_edge_indexes.py 
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
async def provision_edge_indexes(
    database: Any,
    *,
    collection_prefix: str = "graph",
    tenant_ids: Iterable[str] = (),
    schema_writes: bool,
) -> tuple[str, ...]:
    """Apply or verify the two ordered indexes on each tenant edge collection.

    Existing collections are discovered from ``database``. Explicit tenant IDs
    cover new collections before their first edge write. Repeating apply mode
    is safe: MongoDB treats a matching index name and key spec as a no-op.
    Verify mode uses only ``listCollections`` and ``listIndexes`` privileges;
    it fails when a requested tenant has no collection or required index.
    """
    if not collection_prefix:
        raise ValueError("collection_prefix must not be empty")
    prefix = f"{collection_prefix}_"
    suffix = "_edges"
    existing = set(await database.list_collection_names())
    names = {
        name for name in existing if name.startswith(prefix) and name.endswith(suffix)
    }
    for tenant_id in tenant_ids:
        if not tenant_id:
            raise ValueError("tenant_ids must not contain empty values")
        names.add(f"{prefix}{tenant_id}{suffix}")

    for name in sorted(names):
        if not schema_writes and name not in existing:
            raise RuntimeError(f"Missing Mongo edge collection {name}; provision it first")
        collection = database[name]
        found = set()
        if name in existing:
            async for index in collection.list_indexes():
                if not any(index.get(option) for option in (
                    "partialFilterExpression", "sparse", "expireAfterSeconds", "collation",
                    "hidden",
                )):
                    found.add(tuple(index["key"].items()))
        if schema_writes:
            for index_name, keys in _EDGE_INDEXES:
                if keys not in found:
                    await collection.create_index(list(keys), name=index_name)
            continue

        for index_name, keys in _EDGE_INDEXES:
            if keys not in found:
                raise RuntimeError(f"Missing Mongo edge index {name}.{index_name}")
    return tuple(sorted(names))