users

Definition of /auth/users/ routes

list_users(profile_id=None, user=Depends(current_superuser_act), db=Depends(get_async_db)) async

Return list of all users

Source code in fractal_server/app/routes/auth/users.py

@router_users.get("/users/", response_model=list[UserRead])
async def list_users(
    profile_id: int | None = None,
    user: UserOAuth = Depends(current_superuser_act),
    db: AsyncSession = Depends(get_async_db),
):
    """
    Return list of all users
    """
    stm = select(UserOAuth)
    if profile_id is not None:
        stm = stm.where(UserOAuth.profile_id == profile_id)
    res = await db.execute(stm)
    user_list = res.scalars().unique().all()

    # Get all user/group links
    stm_all_links = select(LinkUserGroup)
    res = await db.execute(stm_all_links)
    links = res.scalars().all()

    # TODO: possible optimizations for this construction are listed in
    # https://github.com/fractal-analytics-platform/fractal-server/issues/1742
    for ind, user in enumerate(user_list):
        user_list[ind] = dict(
            **user.model_dump(),
            oauth_accounts=user.oauth_accounts,
            group_ids=[
                link.group_id for link in links if link.user_id == user.id
            ],
        )

    return user_list

patch_user(user_id, user_update, current_superuser=Depends(current_superuser_act), user_manager=Depends(get_user_manager), db=Depends(get_async_db)) async

Custom version of the PATCH-user route from fastapi-users.

Source code in fractal_server/app/routes/auth/users.py

@router_users.patch("/users/{user_id}/", response_model=UserRead)
async def patch_user(
    user_id: int,
    user_update: UserUpdate,
    current_superuser: UserOAuth = Depends(current_superuser_act),
    user_manager: UserManager = Depends(get_user_manager),
    db: AsyncSession = Depends(get_async_db),
):
    """
    Custom version of the PATCH-user route from `fastapi-users`.
    """

    # Check that user exists
    user_to_patch = await _user_or_404(user_id, db)

    if user_update.profile_id is not None:
        profile = await db.get(Profile, user_update.profile_id)
        if profile is None:
            raise HTTPException(
                status_code=status.HTTP_404_NOT_FOUND,
                detail=f"Profile {user_update.profile_id} not found.",
            )

    if user_update.project_dirs is not None:
        await _check_project_dirs_update(
            old_project_dirs=user_to_patch.project_dirs,
            new_project_dirs=user_update.project_dirs,
            user_id=user_id,
            db=db,
        )

    # Modify user attributes
    try:
        user = await user_manager.update(
            user_update,
            user_to_patch,
            safe=False,
            request=None,
        )
        validated_user = UserOAuth.model_validate(user.model_dump())
        patched_user = await db.get(
            UserOAuth, validated_user.id, populate_existing=True
        )
    except exceptions.InvalidPasswordException as e:
        raise HTTPException(
            status_code=status.HTTP_400_BAD_REQUEST,
            detail={
                "code": ErrorCode.UPDATE_USER_INVALID_PASSWORD,
                "reason": e.reason,
            },
        )
    except exceptions.UserAlreadyExists:
        raise HTTPException(
            status.HTTP_400_BAD_REQUEST,
            detail=ErrorCode.UPDATE_USER_EMAIL_ALREADY_EXISTS,
        )

    # Enrich user object with `group_ids_names` attribute
    patched_user_with_groups = await _get_single_user_with_groups(
        patched_user, db
    )

    return patched_user_with_groups