147 lines
3.8 KiB
Python
147 lines
3.8 KiB
Python
# GENERATED BY CLAUDE
|
|
|
|
import os
|
|
import sys
|
|
import uuid
|
|
import asyncio
|
|
from tortoise import Tortoise
|
|
from blueprints.users.models import User
|
|
|
|
from dotenv import load_dotenv
|
|
|
|
load_dotenv()
|
|
|
|
# Database configuration with environment variable support
|
|
DATABASE_PATH = os.getenv("DATABASE_PATH", "database/raggr.db")
|
|
DATABASE_URL = os.getenv("DATABASE_URL", f"sqlite://{DATABASE_PATH}")
|
|
|
|
print(DATABASE_URL)
|
|
|
|
|
|
async def add_user(username: str, email: str, password: str):
|
|
"""Add a new user to the database"""
|
|
await Tortoise.init(
|
|
db_url=DATABASE_URL,
|
|
modules={
|
|
"models": [
|
|
"blueprints.users.models",
|
|
"blueprints.conversation.models",
|
|
]
|
|
},
|
|
)
|
|
|
|
try:
|
|
# Check if user already exists
|
|
existing_user = await User.filter(email=email).first()
|
|
if existing_user:
|
|
print(f"Error: User with email '{email}' already exists!")
|
|
return False
|
|
|
|
existing_username = await User.filter(username=username).first()
|
|
if existing_username:
|
|
print(f"Error: Username '{username}' is already taken!")
|
|
return False
|
|
|
|
# Create new user
|
|
user = User(
|
|
id=uuid.uuid4(),
|
|
username=username,
|
|
email=email,
|
|
)
|
|
user.set_password(password)
|
|
await user.save()
|
|
|
|
print("✓ User created successfully!")
|
|
print(f" Username: {username}")
|
|
print(f" Email: {email}")
|
|
print(f" ID: {user.id}")
|
|
return True
|
|
|
|
except Exception as e:
|
|
print(f"Error creating user: {e}")
|
|
return False
|
|
finally:
|
|
await Tortoise.close_connections()
|
|
|
|
|
|
async def list_users():
|
|
"""List all users in the database"""
|
|
await Tortoise.init(
|
|
db_url=DATABASE_URL,
|
|
modules={
|
|
"models": [
|
|
"blueprints.users.models",
|
|
"blueprints.conversation.models",
|
|
]
|
|
},
|
|
)
|
|
|
|
try:
|
|
users = await User.all()
|
|
if not users:
|
|
print("No users found in database.")
|
|
return
|
|
|
|
print(f"\nFound {len(users)} user(s):")
|
|
print("-" * 60)
|
|
for user in users:
|
|
print(f"Username: {user.username}")
|
|
print(f"Email: {user.email}")
|
|
print(f"ID: {user.id}")
|
|
print(f"Created: {user.created_at}")
|
|
print("-" * 60)
|
|
|
|
except Exception as e:
|
|
print(f"Error listing users: {e}")
|
|
finally:
|
|
await Tortoise.close_connections()
|
|
|
|
|
|
def print_usage():
|
|
"""Print usage instructions"""
|
|
print("Usage:")
|
|
print(" python add_user.py add <username> <email> <password>")
|
|
print(" python add_user.py list")
|
|
print("\nExamples:")
|
|
print(" python add_user.py add ryan ryan@example.com mypassword123")
|
|
print(" python add_user.py list")
|
|
print("\nEnvironment Variables:")
|
|
print(" DATABASE_PATH - Path to database file (default: database/raggr.db)")
|
|
print(" DATABASE_URL - Full database URL (overrides DATABASE_PATH)")
|
|
print("\n Example with custom database:")
|
|
print(" DATABASE_PATH=dev.db python add_user.py list")
|
|
|
|
|
|
async def main():
|
|
if len(sys.argv) < 2:
|
|
print_usage()
|
|
sys.exit(1)
|
|
|
|
command = sys.argv[1].lower()
|
|
|
|
if command == "add":
|
|
if len(sys.argv) != 5:
|
|
print("Error: Missing arguments for 'add' command")
|
|
print_usage()
|
|
sys.exit(1)
|
|
|
|
username = sys.argv[2]
|
|
email = sys.argv[3]
|
|
password = sys.argv[4]
|
|
|
|
success = await add_user(username, email, password)
|
|
sys.exit(0 if success else 1)
|
|
|
|
elif command == "list":
|
|
await list_users()
|
|
sys.exit(0)
|
|
|
|
else:
|
|
print(f"Error: Unknown command '{command}'")
|
|
print_usage()
|
|
sys.exit(1)
|
|
|
|
|
|
if __name__ == "__main__":
|
|
asyncio.run(main())
|