Add cli command for adding users

ghostforge/cli.py

@@ -0,0 +1,71 @@
+import argparse
+import asyncio
+import contextlib
+import getpass
+import re
+from typing import Tuple
+
+from fastapi_users.exceptions import UserAlreadyExists
+
+from ghostforge.db import get_session
+from ghostforge.db import get_user_db
+from ghostforge.users import get_user_manager
+from ghostforge.users import UserCreate
+
+
+get_async_session_context = contextlib.asynccontextmanager(get_session)
+get_user_db_context = contextlib.asynccontextmanager(get_user_db)
+get_user_manager_context = contextlib.asynccontextmanager(get_user_manager)
+
+
+def is_valid_email(email: str) -> bool:
+    email_regex = r"^[a-zA-Z0-9._%+-]+@[a-zA-Z0-9.-]+\.[a-zA-Z]{2,}$"
+    return bool(re.match(email_regex, email))
+
+
+def parse_args() -> Tuple[str, str, str, bool]:
+    parser = argparse.ArgumentParser(description="Create a new user")
+    parser.add_argument("username", type=str, help="Username of the user")
+    parser.add_argument("email", type=str, help="E-mail address of the user")
+    parser.add_argument("--superuser", dest="is_superuser", action="store_true", help="Make user an admin")
+    args = parser.parse_args()
+
+    if not is_valid_email(args.email):
+        parser.error(f"Invalid e-mail address: {args.email}")
+
+    return args.username, args.email, args.is_superuser
+
+
+async def create_user(username: str, email: str, password: str, is_superuser: bool = False):
+    try:
+        async with get_async_session_context() as session:
+            async with get_user_db_context(session) as user_db:
+                async with get_user_manager_context(user_db) as user_manager:
+                    user = await user_manager.create(
+                        UserCreate(email=email, username=username, password=password, is_superuser=is_superuser)
+                    )
+                    if is_superuser:
+                        print(f"Superuser created {user}")
+                    else:
+                        print(f"User created {user}")
+    except UserAlreadyExists:
+        print(f"User {email} already exists")
+
+
+def cli_create_user():
+    username, email, is_superuser = parse_args()
+    password = getpass.getpass(prompt="Password: ")
+
+    if len(password) < 12:
+        print("Error: Password must be at least 12 characters long")
+        return
+
+    if password != getpass.getpass(prompt="Confirm Password: "):
+        print("Error: Passwords do not match")
+        return
+
+    asyncio.run(create_user(username=username, email=email, password=password, is_superuser=is_superuser))
+
+
+if __name__ == "__main__":
+    cli_create_user()

pyproject.toml

@@ -26,6 +26,9 @@ dependencies = [
     "markdown==3.4.3",
 ]
 
+[project.scripts]
+ghostforge_adduser = "ghostforge.cli:cli_create_user"
+
 [project.urls]
 homepage = "https://github.com/DarrylNixon/ghostforge"
 repository = "https://github.com/DarrylNixon/ghostforge"