Skip to content

current_user

Definition of /auth/current-user/ endpoints

get_current_user(group_ids_names=False, user=Depends(current_active_user), db=Depends(get_async_db)) async

Return current user

Source code in fractal_server/app/routes/auth/current_user.py 
29
30
31
32
33
34
35
36
37
38
39
40
41
42
@router_current_user.get("/current-user/", response_model=UserRead)
async def get_current_user(
    group_ids_names: bool = False,
    user: UserOAuth = Depends(current_active_user),
    db: AsyncSession = Depends(get_async_db),
):
    """
    Return current user
    """
    if group_ids_names is True:
        user_with_groups = await _get_single_user_with_groups(user, db)
        return user_with_groups
    else:
        return user

get_current_user_viewer_paths(current_user=Depends(current_active_user), db=Depends(get_async_db)) async

Returns the union of viewer_paths for all user's groups

Source code in fractal_server/app/routes/auth/current_user.py 
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
@router_current_user.get(
    "/current-user/viewer-paths/", response_model=list[str]
)
async def get_current_user_viewer_paths(
    current_user: UserOAuth = Depends(current_active_user),
    db: AsyncSession = Depends(get_async_db),
) -> list[str]:
    """Returns the union of `viewer_paths` for all user's groups"""
    cmd = (
        select(UserGroup.viewer_paths)
        .join(LinkUserGroup)
        .where(LinkUserGroup.group_id == UserGroup.id)
        .where(LinkUserGroup.user_id == current_user.id)
    )
    res = await db.execute(cmd)
    viewer_paths_nested = res.scalars().all()

    # Flatten a nested object and make its elements unique
    all_viewer_paths_set = set(
        path for _viewer_paths in viewer_paths_nested for path in _viewer_paths
    )
    all_viewer_paths = list(all_viewer_paths_set)

    return all_viewer_paths

patch_current_user(user_update, current_user=Depends(current_active_user), user_manager=Depends(get_user_manager), db=Depends(get_async_db)) async

Note: a user cannot patch their own password (as enforced within the UserUpdateStrict schema).

Source code in fractal_server/app/routes/auth/current_user.py 
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
@router_current_user.patch("/current-user/", response_model=UserRead)
async def patch_current_user(
    user_update: UserUpdateStrict,
    current_user: UserOAuth = Depends(current_active_user),
    user_manager: UserManager = Depends(get_user_manager),
    db: AsyncSession = Depends(get_async_db),
):
    """
    Note: a user cannot patch their own password (as enforced within the
    `UserUpdateStrict` schema).
    """
    update = UserUpdate(**user_update.dict(exclude_unset=True))

    # NOTE: here it would be relevant to catch an `InvalidPasswordException`
    # (from `fastapi_users.exceptions`), if we were to allow users change
    # their own password

    user = await user_manager.update(update, current_user, safe=True)
    validated_user = schemas.model_validate(UserOAuth, user)

    patched_user = await db.get(
        UserOAuth, validated_user.id, populate_existing=True
    )
    patched_user_with_groups = await _get_single_user_with_groups(
        patched_user, db
    )
    return patched_user_with_groups