feat: implement admin bootstrap settings and ensure default roles and admin account

services/bootstrap.py

@@ -0,0 +1,129 @@
+from __future__ import annotations
+
+import logging
+from dataclasses import dataclass
+from typing import Callable
+
+from config.settings import AdminBootstrapSettings
+from models import User
+from services.repositories import ensure_default_roles
+from services.unit_of_work import UnitOfWork
+
+
+logger = logging.getLogger(__name__)
+
+
+@dataclass(slots=True)
+class RoleBootstrapResult:
+    created: int
+    ensured: int
+
+
+@dataclass(slots=True)
+class AdminBootstrapResult:
+    created_user: bool
+    updated_user: bool
+    password_rotated: bool
+    roles_granted: int
+
+
+def bootstrap_admin(
+    *,
+    settings: AdminBootstrapSettings,
+    unit_of_work_factory: Callable[[], UnitOfWork] = UnitOfWork,
+) -> tuple[RoleBootstrapResult, AdminBootstrapResult]:
+    """Ensure default roles and administrator account exist."""
+
+    with unit_of_work_factory() as uow:
+        assert uow.roles is not None and uow.users is not None
+
+        role_result = _bootstrap_roles(uow)
+        admin_result = _bootstrap_admin_user(uow, settings)
+
+    logger.info(
+        "Admin bootstrap result: created_user=%s updated_user=%s password_rotated=%s roles_granted=%s",
+        admin_result.created_user,
+        admin_result.updated_user,
+        admin_result.password_rotated,
+        admin_result.roles_granted,
+    )
+    return role_result, admin_result
+
+
+def _bootstrap_roles(uow: UnitOfWork) -> RoleBootstrapResult:
+    assert uow.roles is not None
+    before = {role.name for role in uow.roles.list()}
+    ensure_default_roles(uow.roles)
+    after = {role.name for role in uow.roles.list()}
+    created = len(after - before)
+    return RoleBootstrapResult(created=created, ensured=len(after))
+
+
+def _bootstrap_admin_user(
+    uow: UnitOfWork,
+    settings: AdminBootstrapSettings,
+) -> AdminBootstrapResult:
+    assert uow.users is not None and uow.roles is not None
+
+    created_user = False
+    updated_user = False
+    password_rotated = False
+    roles_granted = 0
+
+    user = uow.users.get_by_email(settings.email, with_roles=True)
+    if user is None:
+        user = User(
+            email=settings.email,
+            username=settings.username,
+            password_hash=User.hash_password(settings.password),
+            is_active=True,
+            is_superuser=True,
+        )
+        uow.users.create(user)
+        created_user = True
+    else:
+        if user.username != settings.username:
+            user.username = settings.username
+            updated_user = True
+        if not user.is_active:
+            user.is_active = True
+            updated_user = True
+        if not user.is_superuser:
+            user.is_superuser = True
+            updated_user = True
+        if settings.force_reset:
+            user.password_hash = User.hash_password(settings.password)
+            password_rotated = True
+            updated_user = True
+        uow.users.session.flush()
+
+    user = uow.users.get(user.id, with_roles=True)
+    assert user is not None
+
+    existing_roles = {role.name for role in user.roles}
+    for role_name in settings.roles:
+        role = uow.roles.get_by_name(role_name)
+        if role is None:
+            logger.warning(
+                "Bootstrap admin role '%s' is not defined; skipping assignment",
+                role_name,
+            )
+            continue
+        if role.name in existing_roles:
+            continue
+        uow.users.assign_role(
+            user_id=user.id,
+            role_id=role.id,
+            granted_by=user.id,
+        )
+        roles_granted += 1
+        existing_roles.add(role.name)
+
+    uow.users.session.flush()
+
+    return AdminBootstrapResult(
+        created_user=created_user,
+        updated_user=updated_user,
+        password_rotated=password_rotated,
+        roles_granted=roles_granted,
+    )