mongo_edge_indexes(database: Annotated[str, Option(help='Mongo database name.')] = 'symfonic', collection_prefix: Annotated[str, Option(help='Graph collection prefix.')] = 'graph', tenant_ids: Annotated[list[str] | None, Option('--tenant-id', help='Also provision a new tenant (repeatable).')] = None, apply: Annotated[bool, Option('--apply', help='Create indexes; default verifies read-only.')] = False) -> None
Verify edge indexes, or create them in an explicit schema-write job.
Reads MONGODB_URI (or MONGO_URL). Apply scans existing tenant edge
collections and creates indexes for them and supplied new tenant IDs.
Run again when another tenant is added. Without --apply no collection
or index is created; missing indexes cause a nonzero exit.
Source code in src/symfonic/cli/mongo_edge_indexes.py
| def mongo_edge_indexes(
database: Annotated[str, typer.Option(help="Mongo database name.")] = "symfonic",
collection_prefix: Annotated[
str, typer.Option(help="Graph collection prefix.")
] = "graph",
tenant_ids: Annotated[
list[str] | None,
typer.Option("--tenant-id", help="Also provision a new tenant (repeatable)."),
] = None,
apply: Annotated[
bool,
typer.Option("--apply", help="Create indexes; default verifies read-only."),
] = False,
) -> None:
"""Verify edge indexes, or create them in an explicit schema-write job.
Reads ``MONGODB_URI`` (or ``MONGO_URL``). Apply scans existing tenant edge
collections and creates indexes for them and supplied new tenant IDs.
Run again when another tenant is added. Without ``--apply`` no collection
or index is created; missing indexes cause a nonzero exit.
"""
uri = os.environ.get("MONGODB_URI") or os.environ.get("MONGO_URL")
if not uri:
typer.echo("MONGODB_URI or MONGO_URL is required", err=True)
raise typer.Exit(2)
try:
import motor.motor_asyncio
from pymongo.errors import PyMongoError
except ImportError as exc:
typer.echo("Install symfonic-core[mongodb] to provision Mongo indexes", err=True)
raise typer.Exit(2) from exc
client = motor.motor_asyncio.AsyncIOMotorClient(uri)
try:
names = asyncio.run(
provision_edge_indexes(
client[database],
collection_prefix=collection_prefix,
tenant_ids=tenant_ids or (),
schema_writes=apply,
)
)
except (RuntimeError, ValueError, PyMongoError) as exc:
typer.echo(str(exc), err=True)
raise typer.Exit(1) from exc
finally:
client.close()
typer.echo(f"{len(names)} tenant edge collections {'provisioned' if apply else 'verified'}")
|