Feature/Create Initial User Command

.github/workflows/ci.yaml

@@ -2,6 +2,9 @@ name: CI
 
 on:
   pull_request:
+  push:
+    branches:
+      - main
   workflow_dispatch:
 
 jobs:

README.md

@@ -8,6 +8,8 @@ Reusable Django utilities and management commands for Toggle projects.
 
 - Shared management command: `wait_for_resources`
   — Wait for database, Redis, Minio (S3) resources to be available before startup
+- Create Initial Users: `create_initial_users`
+    - Create Users with specified roles and permissions, useful to populate the database with default users during development or testing
 
 ---
 
@@ -64,6 +66,23 @@ python manage.py wait_for_resources --db --redis
 ```bash
 python manage.py wait_for_resources --db --redis
 python manage.py wait_for_resources --timeout 300 --minio
+python manage.py create_initial_users --users-json="
+[
+    {
+        "username": "admin",
+        "email": "test@example.com",
+        "password": "admin123",
+        "is_superuser": true,
+        "is_staff": true
+    },
+    {
+        "username": "user1",
+        "email": "user1@gmail.com",
+        "password": "user123",
+        "is_superuser": false,
+        "is_staff": false
+    }
+]'
 ```
 
 ---
@@ -83,6 +102,11 @@ python manage.py wait_for_resources --timeout 300 --minio
     ```bash
     uv run --all-groups --all-extras pytest
     ```
+4. Run commands for example project
+    ```bash
+    uv run --all-groups --all-extras python example/manage.py runserver
+    uv run --all-groups --all-extras python example/manage.py wait_for_resources --db --redis
+    ```
 
 ---
 

example/main/settings.py

@@ -26,7 +26,7 @@
 DATABASES = {
     "default": {
         "ENGINE": "django.db.backends.sqlite3",
-        "NAME": ":memory:",
+        "NAME": BASE_DIR / "db.sqlite3",
     },
 }
 

src/toggle_django_utils/management/commands/create_initial_users.py

@@ -0,0 +1,96 @@
+from __future__ import annotations
+
+import json
+import pathlib
+import typing
+from typing import Any, override
+
+from django.contrib.auth import get_user_model
+from django.contrib.auth.hashers import identify_hasher
+from django.core.management.base import BaseCommand, CommandError
+
+if typing.TYPE_CHECKING:
+    from django.core.management.base import CommandParser
+
+
+class Command(BaseCommand):
+    help = "Create/update users using JSON input"
+
+    @override
+    def add_arguments(self, parser: CommandParser):
+        parser.add_argument(
+            "--users-json",
+            type=str,
+            required=True,
+            help="JSON string containing list of users",
+        )
+
+    def is_hashed(self, password: str) -> bool:
+        try:
+            identify_hasher(password)
+            return True
+        except Exception:
+            return False
+
+    def load_json_string(self, json_str: str) -> list[dict[str, Any]]:
+        try:
+            data = json.loads(json_str)
+            if not isinstance(data, list):
+                raise ValueError("JSON must be a list of users")
+            return data
+        except Exception as e:
+            raise CommandError(f"Invalid JSON: {e}") from e
+
+    def load_json_file(self, path: str) -> list[dict[str, Any]]:
+        try:
+            with pathlib.Path(path).open(encoding="utf-8") as f:
+                data = json.load(f)
+                if not isinstance(data, list):
+                    raise ValueError("JSON must be a list of users")
+                return data
+        except Exception as e:
+            raise CommandError(f"Error reading file {path}: {e}") from e
+
+    @override
+    def handle(self, *_, **options: Any):
+        User = get_user_model()
+        user_data = self.load_json_string(options["users_json"])
+
+        if not user_data:
+            raise CommandError("No users provided")
+
+        for user_info in user_data:
+            email = user_info.pop("email", None)
+            username = user_info.pop("username", email)
+            password = user_info.pop("password", None)
+
+            if not email or not password:
+                self.stdout.write(
+                    self.style.WARNING(
+                        f"Skipping user (missing email/password): {user_info}",
+                    ),
+                )
+                continue
+
+            defaults = {k: v for k, v in user_info.items()}
+            user, created = User.objects.update_or_create(
+                username=username,
+                email=email,
+                defaults=defaults,
+            )
+
+            if self.is_hashed(password):
+                user.password = password
+            else:
+                user.set_password(password)
+
+            user.save()
+
+            if created:
+                self.stdout.write(
+                    self.style.SUCCESS(f"Created user: {email}"),
+                )
+            else:
+                self.stdout.write(
+                    self.style.SUCCESS(f"Updated user: {email}"),
+                )

tests/test_create_initial_users.py

@@ -0,0 +1,59 @@
+import json
+
+import pytest
+from django.contrib.auth import get_user_model
+from django.core.management import call_command
+
+User = get_user_model()
+
+
+@pytest.mark.django_db
+def test_create_users_command():
+    users_json = json.dumps(
+        [
+            {
+                "username": "admin",
+                "email": "admin@example.com",
+                "password": "plainpassword",
+                "is_superuser": True,
+                "is_staff": True,
+            },
+            {
+                "username": "guest",
+                "email": "guest@example.com",
+                "password": "pbkdf2_sha256$600000$example$hashhere",
+                "is_superuser": False,
+                "is_staff": False,
+            },
+            {
+                "first_name": "John",
+                "last_name": "Doe",
+                "email": "john@example.com",
+                "password": "pbkdf2_sha256$600000$example$hashhere",
+                "is_superuser": False,
+                "is_staff": False,
+            },
+        ],
+    )
+
+    call_command("create_initial_users", users_json=users_json)
+
+    admin_user: User = User.objects.get(username="admin")
+    assert admin_user.email == "admin@example.com"  # type: ignore[reportUnknownMemberType]
+    assert admin_user.is_superuser is True  # type: ignore[reportUnknownMemberType]
+    assert admin_user.is_staff is True  # type: ignore[reportUnknownMemberType]
+
+    # password should be hashed, not equal to the plain text
+    assert admin_user.check_password("plainpassword") is True
+
+    guest_user: User = User.objects.get(username="guest")
+    assert guest_user.email == "guest@example.com"  # type: ignore[reportUnknownMemberType]
+    assert guest_user.is_superuser is False  # type: ignore[reportUnknownMemberType]
+    assert guest_user.is_staff is False  # type: ignore[reportUnknownMemberType]
+    assert guest_user.check_password("pbkdf2_sha256$600000$example$hashhere") is False  # type: ignore[reportUnknownMemberType]
+
+    # The user with email as username should be created with email as username
+    john_user: User = User.objects.get(email="john@example.com")
+    assert john_user.username == john_user.email  # type: ignore[reportUnknownMemberType]
+    assert john_user.first_name == "John"  # type: ignore[reportUnknownMemberType]
+    assert john_user.last_name == "Doe"  # type: ignore[reportUnknownMemberType]