user

mlte/backend/api/endpoints/user.py

User CRUD endpoint. Note that all endpoints return a BasicUser instead of a User, which automatically removes the hashed password from the model returned.

create_user(*, user, current_user)

Create a MLTE user.

Parameters:

Name	Type	Description	Default
user	UserWithPassword	The user to create	required

Returns:

Type	Description
BasicUser	The created user

Source code in mlte/backend/api/endpoints/user.py

@router.post("")
def create_user(
    *,
    user: UserWithPassword,
    current_user: AuthorizedUser,
) -> BasicUser:
    """
    Create a MLTE user.
    :param user: The user to create
    :return: The created user
    """
    if user.username == USER_ME_ID:
        raise HTTPException(
            status_code=codes.BAD_REQUEST,
            detail="'me' is reserved and can't be used as a username.",
        )

    with state_stores.user_store_session() as user_store:
        try:
            # Give every new user permissions to create (only) new models.
            model_create_policy = Policy(
                ResourceType.MODEL,
                resource_id=None,
                create_group=True,
                edit_group=False,
                read_group=False,
            )
            model_create_policy.assign_to_user(user)

            # Give every new user permissions to modify all custom lists
            custom_list_policy = Policy(
                ResourceType.CUSTOM_LIST, resource_id=None
            )
            custom_list_policy.assign_to_user(user)

            # Give user permissions to modify its data.
            own_user_policy = Policy(
                ResourceType.USER, resource_id=user.username
            )
            own_user_policy.save_to_store(user_store)
            own_user_policy.assign_to_user(user)

            # Store the user.
            new_user: BasicUser = user_store.user_mapper.create(user)
            stored_user = user_store.user_mapper.read(new_user.username)
            return stored_user
        except errors.ErrorAlreadyExists as e:
            raise HTTPException(
                status_code=codes.ALREADY_EXISTS, detail=f"{e} already exists."
            )
        except Exception as e:
            raise_http_internal_error(e)

delete_user(*, username, current_user)

Delete a MLTE user.

Parameters:

Name	Type	Description	Default
username	str	The username	required

Returns:

Type	Description
BasicUser	The deleted user

Source code in mlte/backend/api/endpoints/user.py

@router.delete("/{username}")
def delete_user(
    *,
    username: str,
    current_user: AuthorizedUser,
) -> BasicUser:
    """
    Delete a MLTE user.
    :param username: The username
    :return: The deleted user
    """
    with state_stores.user_store_session() as user_store:
        try:
            deleted_user = user_store.user_mapper.delete(username)

            # Now delete related permissions and groups.
            policy = Policy(ResourceType.USER, resource_id=username)
            policy.remove_from_store(user_store)

            return deleted_user
        except errors.ErrorNotFound as e:
            raise HTTPException(
                status_code=codes.NOT_FOUND, detail=f"{e} not found."
            )
        except Exception as e:
            raise_http_internal_error(e)

edit_user(*, user, current_user)

Edit a MLTE user.

Parameters:

Name	Type	Description	Default
user	Union[UserWithPassword, BasicUser]	The user to edit	required

Returns:

Type	Description
BasicUser	The edited user

Source code in mlte/backend/api/endpoints/user.py

@router.put("")
def edit_user(
    *,
    user: Union[UserWithPassword, BasicUser],
    current_user: AuthorizedUser,
) -> BasicUser:
    """
    Edit a MLTE user.
    :param user: The user to edit
    :return: The edited user
    """
    if user.username == USER_ME_ID:
        user.username = current_user.username

    with state_stores.user_store_session() as user_store:
        try:
            # We only want to allow admins to edit a user's groups.
            if current_user.role != RoleType.ADMIN:
                # If not admin, keep current groups and ignore the new ones, if any.
                current_groups = user_store.user_mapper.read(
                    user.username
                ).groups
                user.groups = current_groups

            # Edit the user.
            return user_store.user_mapper.edit(user)
        except errors.ErrorNotFound as e:
            raise HTTPException(
                status_code=codes.NOT_FOUND, detail=f"{e} not found."
            )
        except Exception as e:
            raise_http_internal_error(e)

list_user_models(*, username, current_user)

Gets a list of models a user is authorized to read.

Parameters:

Name	Type	Description	Default
username	str	The username	required

Returns:

Type	Description
List[str]	The list of model ids

Source code in mlte/backend/api/endpoints/user.py

@router.get("/{username}/models")
def list_user_models(
    *,
    username: str,
    current_user: AuthorizedUser,
) -> List[str]:
    """
    Gets a list of models a user is authorized to read.
    :param username: The username
    :return: The list of model ids
    """
    if username == USER_ME_ID:
        username = current_user.username

    with state_stores.artifact_store_session() as artifact_store:
        with state_stores.user_store_session() as user_store:
            try:
                # Get all models, and filter out only the ones the user has read permissions for.
                user_models: List[str] = []
                user = BasicUser(
                    **user_store.user_mapper.read(username).to_json()
                )
                all_models = artifact_store.list_models()
                for model_id in all_models:
                    permission = Permission(
                        resource_type=ResourceType.MODEL,
                        resource_id=model_id,
                        method=MethodType.GET,
                    )
                    if authorization.is_authorized(user, permission):
                        user_models.append(model_id)
                return user_models

            except errors.ErrorNotFound as e:
                raise HTTPException(
                    status_code=codes.NOT_FOUND, detail=f"{e} not found."
                )
            except Exception as e:
                raise_http_internal_error(e)

list_user_models_me(*, current_user)

Gets a list of models the currently logged-in user is authorized to read.

Returns:

Type	Description
List[str]	The list of model ids

Source code in mlte/backend/api/endpoints/user.py

@router.get("/me/models")
def list_user_models_me(
    *,
    current_user: AuthorizedUser,
) -> List[str]:
    """
    Gets a list of models the currently logged-in user is authorized to read.
    :return: The list of model ids
    """
    parameters = locals().copy()
    parameters["username"] = USER_ME_ID
    return list_user_models(**parameters)

list_users(current_user)

List MLTE users.

Returns:

Type	Description
List[str]	A collection of usernames

Source code in mlte/backend/api/endpoints/user.py

@router.get("")
def list_users(
    current_user: AuthorizedUser,
) -> List[str]:
    """
    List MLTE users.
    :return: A collection of usernames
    """
    with state_stores.user_store_session() as user_store:
        try:
            return user_store.user_mapper.list()
        except Exception as e:
            raise_http_internal_error(e)

list_users_details(current_user)

List MLTE users, with details for each user.

Returns:

Type	Description
List[BasicUser]	A collection of users with their details.

Source code in mlte/backend/api/endpoints/user.py

@router.get("s/details")
def list_users_details(
    current_user: AuthorizedUser,
) -> List[BasicUser]:
    """
    List MLTE users, with details for each user.
    :return: A collection of users with their details.
    """
    with state_stores.user_store_session() as user_store:
        try:
            detailed_users = []
            usernames = user_store.user_mapper.list()
            for username in usernames:
                user_details = BasicUser(
                    **user_store.user_mapper.read(username).to_json()
                )
                detailed_users.append(user_details)
            return detailed_users
        except Exception as e:
            raise_http_internal_error(e)

read_user(*, username, current_user)

Read a MLTE user.

Parameters:

Name	Type	Description	Default
username	str	The username	required

Returns:

Type	Description
BasicUser	The read user

Source code in mlte/backend/api/endpoints/user.py

@router.get("/{username}")
def read_user(
    *,
    username: str,
    current_user: AuthorizedUser,
) -> BasicUser:
    """
    Read a MLTE user.
    :param username: The username
    :return: The read user
    """
    if username == USER_ME_ID:
        return current_user

    with state_stores.user_store_session() as user_store:
        try:
            return user_store.user_mapper.read(username)
        except errors.ErrorNotFound as e:
            raise HTTPException(
                status_code=codes.NOT_FOUND, detail=f"{e} not found."
            )
        except Exception as e:
            raise_http_internal_error(e)

read_user_me(current_user)

Returns the currently logged in user.

Returns:

Type	Description
BasicUser	The user info

Source code in mlte/backend/api/endpoints/user.py

@router.get("/me")
def read_user_me(
    current_user: AuthorizedUser,
) -> BasicUser:
    """
    Returns the currently logged in user.
    :return: The user info
    """
    parameters = locals().copy()
    parameters["username"] = USER_ME_ID
    return read_user(**parameters)