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()