zzstoatzz.io
audio streaming app
plyr.fm
1#!/usr/bin/env python
2"""admin script to manage feature flags for users.
3
4usage (from repo root):
5 cd backend && uv run python ../scripts/feature_flag.py enable --user <did_or_handle> --flag <flag_name>
6 cd backend && uv run python ../scripts/feature_flag.py disable --user <did_or_handle> --flag <flag_name>
7 cd backend && uv run python ../scripts/feature_flag.py list --user <did_or_handle>
8 cd backend && uv run python ../scripts/feature_flag.py list-all
9
10environment variables:
11 DATABASE_URL - database connection string
12
13examples:
14 # enable lossless uploads for a user
15 cd backend && DATABASE_URL="..." uv run python ../scripts/feature_flag.py enable --user did:plc:abc123 --flag lossless-uploads
16
17 # disable a flag
18 cd backend && DATABASE_URL="..." uv run python ../scripts/feature_flag.py disable --user zzstoatzz.io --flag lossless-uploads
19
20 # list flags for a user
21 cd backend && DATABASE_URL="..." uv run python ../scripts/feature_flag.py list --user zzstoatzz.io
22
23 # list all users with flags
24 cd backend && DATABASE_URL="..." uv run python ../scripts/feature_flag.py list-all
25"""
26
27import argparse
28import asyncio
29import os
30import sys
31from pathlib import Path
32
33# add backend/src to path
34sys.path.insert(0, str(Path(__file__).parent.parent / "backend" / "src"))
35
36
37def get_database_url() -> str:
38 """get database URL from environment."""
39 url = os.environ.get("DATABASE_URL")
40 if not url:
41 print("error: DATABASE_URL required")
42 print("set DATABASE_URL to your database connection string")
43 sys.exit(1)
44 return url
45
46
47async def resolve_user(db, did_or_handle: str):
48 """resolve a DID or handle to an Artist record."""
49 from sqlalchemy import select
50
51 from backend.models import Artist
52
53 # check if it's a DID
54 if did_or_handle.startswith("did:"):
55 result = await db.execute(select(Artist).where(Artist.did == did_or_handle))
56 else:
57 # treat as handle
58 result = await db.execute(select(Artist).where(Artist.handle == did_or_handle))
59
60 return result.scalar_one_or_none()
61
62
63async def cmd_enable(args) -> None:
64 """enable a feature flag for a user."""
65 from backend._internal import enable_flag, get_user_flags
66 from backend.utilities.database import db_session
67
68 async with db_session() as db:
69 artist = await resolve_user(db, args.user)
70 if not artist:
71 print(f"error: user not found: {args.user}")
72 sys.exit(1)
73
74 newly_enabled = await enable_flag(db, artist.did, args.flag)
75 await db.commit()
76
77 if newly_enabled:
78 print(f"enabled '{args.flag}' for {artist.handle} ({artist.did})")
79 else:
80 print(f"flag '{args.flag}' already enabled for {artist.handle}")
81
82 flags = await get_user_flags(db, artist.did)
83 print(f"flags: {flags}")
84
85
86async def cmd_disable(args) -> None:
87 """disable a feature flag for a user."""
88 from backend._internal import disable_flag, get_user_flags
89 from backend.utilities.database import db_session
90
91 async with db_session() as db:
92 artist = await resolve_user(db, args.user)
93 if not artist:
94 print(f"error: user not found: {args.user}")
95 sys.exit(1)
96
97 was_disabled = await disable_flag(db, artist.did, args.flag)
98 await db.commit()
99
100 if was_disabled:
101 print(f"disabled '{args.flag}' for {artist.handle} ({artist.did})")
102 else:
103 print(f"flag '{args.flag}' not enabled for {artist.handle}")
104
105 flags = await get_user_flags(db, artist.did)
106 print(f"flags: {flags}")
107
108
109async def cmd_list(args) -> None:
110 """list flags for a user."""
111 from backend._internal import get_user_flags
112 from backend.utilities.database import db_session
113
114 async with db_session() as db:
115 artist = await resolve_user(db, args.user)
116 if not artist:
117 print(f"error: user not found: {args.user}")
118 sys.exit(1)
119
120 flags = await get_user_flags(db, artist.did)
121 print(f"{artist.handle} ({artist.did}):")
122 if flags:
123 for flag in flags:
124 print(f" - {flag}")
125 else:
126 print(" (no flags enabled)")
127
128
129async def cmd_list_all(args) -> None:
130 """list all users with feature flags."""
131 from sqlalchemy import select
132
133 from backend.models import Artist, FeatureFlag
134 from backend.utilities.database import db_session
135
136 async with db_session() as db:
137 # find all unique user DIDs with flags
138 result = await db.execute(select(FeatureFlag.user_did).distinct())
139 dids_with_flags = list(result.scalars().all())
140
141 if not dids_with_flags:
142 print("no users have feature flags enabled")
143 return
144
145 # get artist info for each DID
146 artist_result = await db.execute(
147 select(Artist).where(Artist.did.in_(dids_with_flags))
148 )
149 artists_by_did = {a.did: a for a in artist_result.scalars().all()}
150
151 # get all flags grouped by user
152 flags_result = await db.execute(select(FeatureFlag))
153 all_flags = flags_result.scalars().all()
154
155 # group flags by DID
156 flags_by_did: dict[str, list[str]] = {}
157 for flag in all_flags:
158 flags_by_did.setdefault(flag.user_did, []).append(flag.flag)
159
160 print(f"users with feature flags ({len(dids_with_flags)}):")
161 for did in dids_with_flags:
162 artist = artists_by_did.get(did)
163 handle = artist.handle if artist else "(unknown)"
164 flags = flags_by_did.get(did, [])
165 print(f"\n{handle} ({did}):")
166 for flag in flags:
167 print(f" - {flag}")
168
169
170def main() -> None:
171 """main entry point."""
172 parser = argparse.ArgumentParser(description="manage feature flags")
173 subparsers = parser.add_subparsers(dest="command", required=True)
174
175 # enable command
176 enable_parser = subparsers.add_parser("enable", help="enable a flag for a user")
177 enable_parser.add_argument("--user", required=True, help="user DID or handle")
178 enable_parser.add_argument(
179 "--flag", required=True, help="flag name (e.g., lossless-uploads)"
180 )
181
182 # disable command
183 disable_parser = subparsers.add_parser("disable", help="disable a flag for a user")
184 disable_parser.add_argument("--user", required=True, help="user DID or handle")
185 disable_parser.add_argument(
186 "--flag", required=True, help="flag name (e.g., lossless-uploads)"
187 )
188
189 # list command
190 list_parser = subparsers.add_parser("list", help="list flags for a user")
191 list_parser.add_argument("--user", required=True, help="user DID or handle")
192
193 # list-all command
194 subparsers.add_parser("list-all", help="list all users with flags")
195
196 args = parser.parse_args()
197
198 # setup database URL
199 os.environ["DATABASE_URL"] = get_database_url()
200
201 # run command
202 if args.command == "enable":
203 asyncio.run(cmd_enable(args))
204 elif args.command == "disable":
205 asyncio.run(cmd_disable(args))
206 elif args.command == "list":
207 asyncio.run(cmd_list(args))
208 elif args.command == "list-all":
209 asyncio.run(cmd_list_all(args))
210
211
212if __name__ == "__main__":
213 main()