File size: 2,469 Bytes
824d0dd | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 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 | #!/usr/bin/env python
"""Initialize a D2 event tenant (and optionally the first operator invitation).
Trusted operator CLI. Requires an EXPLICIT confirmed-local D2 database path. Prints any
generated invitation code exactly once; never prints secret configuration.
Usage:
python scripts/d2_initialize_tenant.py --db-path /tmp/amanpay-d2/amanpay.db \
--slug event-alpha --name-en "Alpha Event" --name-ar "فعالية ألفا" [--operator-invite]
The database is created fresh only with --initialize (single-use bootstrap).
"""
from __future__ import annotations
import argparse
from amanpay.identity.cli import emit, open_storage, snapshot_best_effort
from amanpay.identity.repository import D2Repository
from amanpay.identity.service import IdentityService
def main(argv=None) -> int:
ap = argparse.ArgumentParser(description="Initialize a D2 event tenant")
ap.add_argument("--db-path", required=True)
ap.add_argument("--slug", required=True)
ap.add_argument("--name-en", default="")
ap.add_argument("--name-ar", default="")
ap.add_argument("--initialize", action="store_true",
help="create a fresh empty DB if none exists (single-use bootstrap)")
ap.add_argument("--operator-invite", action="store_true",
help="also issue a one-time operator invitation and print its code once")
ap.add_argument("--invite-ttl", type=int, default=86400)
args = ap.parse_args(argv)
storage = open_storage(args.db_path, init=args.initialize)
try:
repo = D2Repository(storage.db)
existing = repo.get_tenant_by_slug(args.slug)
tenant = existing or repo.create_tenant(
slug=args.slug, name_en=args.name_en, name_ar=args.name_ar)
out = {"tenant_id": tenant.id, "slug": tenant.slug, "status": tenant.status,
"created": existing is None}
if args.operator_invite:
ident = IdentityService.from_storage(storage)
inv, code = ident.invites.create(
tenant_id=tenant.id, role="operator", ttl_seconds=args.invite_ttl)
out["operator_invitation"] = {"id": inv.id, "code": code,
"note": "Shown once — not stored in plaintext."}
gen = snapshot_best_effort(storage)
out["snapshot_generation"] = gen
emit(out)
finally:
storage.close()
return 0
if __name__ == "__main__":
raise SystemExit(main())
|