Skip to content

user

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
User

The created user

Source code in mlte/backend/api/endpoints/user.py
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
@router.post("")
def create_user(
    *,
    user: UserWithPassword,
    current_user: AuthorizedUser,
) -> User:
    """
    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:
            new_user = 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
User

The deleted user

Source code in mlte/backend/api/endpoints/user.py
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
@router.delete("/{username}")
def delete_user(
    *,
    username: str,
    current_user: AuthorizedUser,
) -> User:
    """
    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)
            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
User

The edited user

Source code in mlte/backend/api/endpoints/user.py
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
@router.put("")
def edit_user(
    *,
    user: Union[UserWithPassword, BasicUser],
    current_user: AuthorizedUser,
) -> User:
    """
    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:
            # TODO: this is a weird permission check that is outside the common permission checks.
            # We only want to allow admins to edit a user's groups.
            if current_user.role != RoleType.ADMIN:
                user = user_policy.remove_new_groups(user, user_store)

            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
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
@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.model_mapper.list()
                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
52
53
54
55
56
57
58
59
60
61
62
63
@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
157
158
159
160
161
162
163
164
165
166
167
168
169
@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[User]

A collection of users with their details.

Source code in mlte/backend/api/endpoints/user.py
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
@router.get("s/details")
def list_users_details(
    current_user: AuthorizedUser,
) -> List[User]:
    """
    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 = User(
                    **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
User

The read user

Source code in mlte/backend/api/endpoints/user.py
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
@router.get("/{username}")
def read_user(
    *,
    username: str,
    current_user: AuthorizedUser,
) -> User:
    """
    Read a MLTE user.
    :param username: The username
    :return: The read user
    """
    # For the special USER_ME_ID, get the current user's username.
    if username == USER_ME_ID:
        username = current_user.username

    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
39
40
41
42
43
44
45
46
47
48
49
@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)