Spaces:
Running
Running
File size: 2,921 Bytes
2d0d464 | 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 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 | """
Make a user permanent (is_permanent = True).
Uses the app's SQLAlchemy models so it matches the app exactly. Targets by email,
prints the matched record BEFORE updating, and asks for confirmation.
Usage (from backend/):
python scripts/make_user_permanent.py gourivb842@gmail.com
python scripts/make_user_permanent.py gourivb842@gmail.com --yes # skip prompt
Needs the same env as the app (POSTGRES_URI / DATABASE_URL, JWT_SECRET, etc.),
e.g. load your .env first or run with the vars exported.
"""
import argparse
import asyncio
import sys
from pathlib import Path
# Ensure "app" imports work when running this file directly.
BACKEND_ROOT = Path(__file__).resolve().parent.parent
if str(BACKEND_ROOT) not in sys.path:
sys.path.insert(0, str(BACKEND_ROOT))
from sqlalchemy import select
from app.database import close_db, connect_db, get_session
from app.models import User
async def _run(email: str, assume_yes: bool) -> int:
await connect_db()
try:
async with get_session() as session:
matches = (
(await session.execute(select(User).where(User.email == email)))
.scalars()
.all()
)
if not matches:
print(f"No user found with email {email!r}.")
return 1
if len(matches) > 1:
print(f"WARNING: {len(matches)} records share email {email!r}:")
for i, u in enumerate(matches):
print(
f" [{i}] id={u.id} role={u.role!r} "
f"is_permanent={u.is_permanent!r} supabase_uid={u.supabase_uid!r}"
)
if len(matches) > 1:
print(
"\nMultiple rows matched. Resolve the duplicate first. "
"Not updating to stay safe."
)
return 2
user = matches[0]
if user.is_permanent is True:
print("Already permanent. Nothing to do.")
return 0
if not assume_yes:
ans = (
input(f"Set is_permanent=True for {email} (id={user.id})? [y/N] ")
.strip()
.lower()
)
if ans not in ("y", "yes"):
print("Aborted.")
return 0
user.is_permanent = True
print(f"{email} is now permanent.")
return 0
finally:
await close_db()
def main() -> None:
parser = argparse.ArgumentParser(description="Make a user permanent.")
parser.add_argument("email", help="email of the user to make permanent")
parser.add_argument(
"--yes", action="store_true", help="skip the confirmation prompt"
)
args = parser.parse_args()
sys.exit(asyncio.run(_run(args.email, args.yes)))
if __name__ == "__main__":
main()
|