Add cli commands for user management
This commit is contained in:
79
mft/cli.py
79
mft/cli.py
@@ -36,16 +36,19 @@ def parse_args() -> argparse.Namespace:
|
||||
"--debug", action="store_true", help="Run in debug mode with auto-reload"
|
||||
)
|
||||
|
||||
# db subcommand with nested operations
|
||||
# db (init, backup) subcommands
|
||||
db_parser = subparsers.add_parser("db", help="Database operations")
|
||||
db_subparsers = db_parser.add_subparsers(dest="db_command", required=True)
|
||||
|
||||
# db init
|
||||
db_init_parser = db_subparsers.add_parser("init", help="Initialize database")
|
||||
|
||||
# db backup
|
||||
db_backup_parser = db_subparsers.add_parser("backup", help="Create database backup")
|
||||
|
||||
# user (list, add) subcommands
|
||||
user_parser = subparsers.add_parser("user", help="User management")
|
||||
user_subparsers = user_parser.add_subparsers(dest="user_command", required=True)
|
||||
user_list_parser = user_subparsers.add_parser("list", help="List all users")
|
||||
user_add_parser = user_subparsers.add_parser("add", help="Add a new user")
|
||||
user_add_parser.add_argument("name", type=str, help="User name")
|
||||
|
||||
return transform_args(parser.parse_args())
|
||||
|
||||
|
||||
@@ -73,6 +76,8 @@ def main():
|
||||
run_command(args, settings)
|
||||
elif args.command == "db":
|
||||
db_command(args, settings)
|
||||
elif args.command == "user":
|
||||
user_command(args, settings)
|
||||
|
||||
|
||||
def run_command(args, settings):
|
||||
@@ -138,5 +143,69 @@ def db_backup_command(args, settings):
|
||||
sys.exit(1)
|
||||
|
||||
|
||||
def user_command(args, settings):
|
||||
if args.user_command == "list":
|
||||
user_list_command(args, settings)
|
||||
elif args.user_command == "add":
|
||||
user_add_command(args, settings)
|
||||
|
||||
|
||||
def user_list_command(args, settings):
|
||||
from mft.database import get_db
|
||||
|
||||
with get_db() as conn:
|
||||
cursor = conn.cursor()
|
||||
# Get all users with count of active tokens
|
||||
cursor.execute(
|
||||
"""
|
||||
SELECT
|
||||
u.id,
|
||||
u.name,
|
||||
COUNT(CASE WHEN a.enabled = 1 THEN 1 END) as active_tokens
|
||||
FROM user u
|
||||
LEFT JOIN auth a ON u.id = a.uid
|
||||
GROUP BY u.id, u.name
|
||||
ORDER BY u.id
|
||||
"""
|
||||
)
|
||||
users = cursor.fetchall()
|
||||
|
||||
if not users:
|
||||
print("No users found.")
|
||||
return
|
||||
|
||||
print(f"{'ID':<5} {'Name':<20} {'Active Tokens':<15}")
|
||||
print("-" * 40)
|
||||
|
||||
for user in users:
|
||||
user_id = user[0]
|
||||
name = user[1]
|
||||
active_tokens = user[2]
|
||||
print(f"{user_id:<5} {name:<20} {active_tokens:<15}")
|
||||
|
||||
print(f"\nTotal users: {len(users)}")
|
||||
|
||||
|
||||
def user_add_command(args, settings):
|
||||
from mft.database import get_db
|
||||
|
||||
with get_db() as conn:
|
||||
cursor = conn.cursor()
|
||||
try:
|
||||
cursor.execute("INSERT INTO user (name) VALUES (?)", (args.name,))
|
||||
conn.commit()
|
||||
user_id = cursor.lastrowid
|
||||
print(f"User '{args.name}' created successfully with ID: {user_id}")
|
||||
except sqlite3.IntegrityError as e:
|
||||
if "UNIQUE constraint failed" in str(e):
|
||||
print(
|
||||
f"Error: User with name '{args.name}' already exists.",
|
||||
file=sys.stderr,
|
||||
)
|
||||
else:
|
||||
print(f"Error: {e}", file=sys.stderr)
|
||||
sys.exit(1)
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
main()
|
||||
|
||||
Reference in New Issue
Block a user