agent-platform/services/auth-service/app/bootstrap/demo_seed.py

import logging

from sqlalchemy.exc import SQLAlchemyError
from sqlalchemy.orm import sessionmaker

from app.bootstrap.settings import AuthServiceSettings
from app.domain.repositories import (
    RoleAssignmentRepository,
    RolePermissionBindingRepository,
    RoleRepository,
    UserRepository,
)
from app.infrastructure.passwords import hash_password

logger = logging.getLogger(__name__)


def bootstrap_demo_identity(
    *,
    settings: AuthServiceSettings,
    session_factory: sessionmaker) -> None:
    if not settings.demo_user_bootstrap_enabled or settings.service_env != "local":
        return

    db = session_factory()
    try:
        users = UserRepository(db)
        if users.has_any():
            return

        user = users.create(
            username=settings.demo_user_username,
            password_hash=hash_password(settings.demo_user_password),
            display_name=settings.demo_user_display_name,
            email=settings.demo_user_email,
            metadata_json={"source": "local-bootstrap"})

        roles = RoleRepository(db)
        role = roles.get_by_name(name="Administrator")
        if role is None:
            role = roles.create(
                code="administrator",
                name="Administrator",
                description="Local bootstrap administrator",
                permissions_json=[])

        RoleAssignmentRepository(db).create(
            user_id=user.id,
            role_id=role.id,
            scope_type=None,
            scope_id=None,
            expires_time=None)
        RolePermissionBindingRepository(db).create(
            role_id=role.id,
            permission="*",
            scope_type=None,
            scope_id=None)
    except SQLAlchemyError as exc:
        db.rollback()
        logger.warning("Skipped demo identity bootstrap: %s", exc)
    finally:
        db.close()