Skip to content

users

ArkIdentityUsersService

Bases: ArkIdentityBaseService

Source code in ark_sdk_python/services/identity/users/ark_identity_users_service.py 
 43
 44
 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
 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
 99
100
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
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
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
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
289
290
291
292
293
294
295
296
297
298
299
300
301
302
303
304
305
306
307
308
309
310
311
312
313
314
315
316
317
318
319
320
321
322
323
324
325
326
327
328
329
330
331
332
333
334
class ArkIdentityUsersService(ArkIdentityBaseService):
    def create_user(self, create_user: ArkIdentityCreateUser) -> ArkIdentityUser:
        """
        Creates a user with the given details, and returns its finalized details and id

        Args:
            create_user (ArkIdentityCreateUser): _description_

        Raises:
            ArkServiceException: _description_

        Returns:
            ArkIdentityUser: _description_
        """
        self._logger.info(f'Creating identity user [{create_user.username}]')
        directories_service = ArkIdentityDirectoriesService(self._isp_auth)
        tenant_suffix = create_user.suffix or directories_service.tenant_default_suffix()
        response: Response = self._client.post(
            f'{self._url_prefix}{CREATE_USER_URL}',
            json={
                "DisplayName": create_user.display_name,
                "Name": f'{create_user.username}@{tenant_suffix}',
                "Mail": create_user.email,
                "Password": create_user.password.get_secret_value(),
                "MobileNumber": create_user.mobile_number,
                "InEverybodyRole": 'true',
                "InSysAdminRole": 'false',
                "ForcePasswordChangeNext": 'false',
                "SendEmailInvite": 'false',
                "SendSmsInvite": 'false',
            },
        )
        try:
            result = response.json()
            if response.status_code != HTTPStatus.OK or not result['success']:
                raise ArkServiceException(f'Failed to create user [{response.text}]')
            if create_user.roles:
                roles_service = ArkIdentityRolesService(self._isp_auth)
                for role in create_user.roles:
                    roles_service.add_user_to_role(
                        ArkIdentityAddUserToRole(username=f'{create_user.username}@{tenant_suffix}', role_name=role)
                    )
            self._logger.info(f'User created successfully with id [{result["Result"]}]')
            return ArkIdentityUser(
                user_id=result['Result'],
                username=f'{create_user.username}@{tenant_suffix}',
                display_name=create_user.display_name,
                email=create_user.email,
                mobile_number=create_user.mobile_number,
                roles=create_user.roles,
            )
        except (ValidationError, JSONDecodeError, KeyError) as ex:
            self._logger.exception(f'Failed to parse create user response [{str(ex)}] - [{response.text}]')
            raise ArkServiceException(f'Failed to parse create user response [{str(ex)}]') from ex

    def update_user(self, update_user: ArkIdentityUpdateUser) -> None:
        """
        Updates the user information

        Args:
            update_user (ArkIdentityUpdateUser): _description_

        Raises:
            ArkServiceException: _description_
        """
        if update_user.username and not update_user.user_id:
            update_user.user_id = self.user_id_by_name(ArkIdentityUserIdByName(username=update_user.username))
        self._logger.info(f'Updating identity user [{update_user.user_id}]')
        update_dict = {}
        if update_user.new_username:
            if '@' not in update_user.new_username:
                tenant_suffix = update_user.username.split('@')[1]
                update_user.new_username = f'{update_user.new_username}@{tenant_suffix}'
            update_dict['Name'] = update_user.new_username
        if update_user.display_name:
            update_dict['DisplayName'] = update_user.display_name
        if update_user.email:
            update_dict['Mail'] = update_user.email
        if update_user.mobile_number:
            update_dict['MobileNumber'] = update_user.mobile_number
        update_dict['ID'] = update_user.user_id
        response: Response = self._client.post(f'{self._url_prefix}{UPDATE_USER_URL}', json=update_dict)
        try:
            result = response.json()
            if response.status_code != HTTPStatus.OK or not result['success']:
                raise ArkServiceException(f'Failed to update user [{response.text}]')
            self._logger.info('User updated successfully')
        except (ValidationError, JSONDecodeError, KeyError) as ex:
            self._logger.exception(f'Failed to parse update user response [{str(ex)}] - [{response.text}]')
            raise ArkServiceException(f'Failed to parse update user response [{str(ex)}]') from ex

    def delete_user(self, delete_user: ArkIdentityDeleteUser) -> None:
        """
        Deletes a user by given name

        Args:
            delete_user (ArkIdentityDeleteUser): _description_

        Raises:
            ArkServiceException: _description_
        """
        self._logger.info(f'Deleting user [{delete_user.user_id}]')
        response: Response = self._client.post(
            f'{self._url_prefix}{DELETE_USER_URL}', json={'ID': delete_user.user_id or delete_user.username}
        )
        try:
            if response.status_code != HTTPStatus.OK or not response.json()['success']:
                raise ArkServiceException(f'Failed to delete user [{response.text}]')
        except (ValidationError, JSONDecodeError, KeyError) as ex:
            self._logger.exception(f'Failed to parse delete user response [{str(ex)}] - [{response.text}]')
            raise ArkServiceException(f'Failed to parse delete user response [{str(ex)}]') from ex

    def delete_users(self, delete_users: ArkIdentityDeleteUsers) -> None:
        """
        Deletes users by given ids

        Args:
            delete_users (ArkIdentityDeleteUsers): _description_

        Raises:
            ArkServiceException: _description_
        """
        self._logger.info(f'Removing users [{",".join(delete_users.user_ids)}]')
        response: Response = self._client.post(f'{self._url_prefix}{REMOVE_USERS_URL}', json={'Users': [delete_users.user_ids]})
        try:
            if response.status_code != HTTPStatus.OK or not response.json()['success']:
                raise ArkServiceException(f'Failed to remove users [{response.text}]')
        except (ValidationError, JSONDecodeError, KeyError) as ex:
            self._logger.exception(f'Failed to parse remove users response [{str(ex)}] - [{response.text}]')
            raise ArkServiceException(f'Failed to parse remove users response [{str(ex)}]') from ex

    def user_id_by_name(self, user_id_by_name: ArkIdentityUserIdByName) -> str:
        """
        Finds the identifier of the given username

        Args:
            user_id_by_name (ArkIdentityUserIdByName): _description_

        Returns:
            str: _description_
        """
        response: Response = self._client.post(
            f'{self._url_prefix}{REDROCK_QUERY}',
            json={"Script": f"Select ID, Username from User WHERE Username='{user_id_by_name.username}'"},
        )
        if response.status_code != HTTPStatus.OK:
            raise ArkServiceException(f'Failed to retrieve user id by name [{response.text}] - [{response.status_code}]')
        try:
            query_result = response.json()
            if not query_result['success'] or len(query_result['Result']["Results"]) == 0:
                raise ArkServiceException('Failed to retrieve user id by name')
            return query_result['Result']["Results"][0]["Row"]["ID"]
        except (ValidationError, JSONDecodeError, KeyError) as ex:
            self._logger.exception(f'Failed to parse user id by name response [{str(ex)}] - [{response.text}]')
            raise ArkServiceException(f'Failed to parse user id by name response [{str(ex)}]') from ex

    def user_by_name(self, user_id_by_name: ArkIdentityUserByName) -> ArkIdentityUser:
        """
        Finds the identifier of the given username

        Args:
            user_id_by_name (ArkIdentityUserIdByName): _description_

        Returns:
            str: _description_
        """
        response: Response = self._client.post(
            f'{self._url_prefix}{REDROCK_QUERY}',
            json={
                "Script": f"Select ID, Username, DisplayName, Email, MobileNumber, LastLogin from User WHERE Username='{user_id_by_name.username}'"
            },
        )
        if response.status_code != HTTPStatus.OK:
            raise ArkServiceException(f'Failed to retrieve user id by name [{response.text}] - [{response.status_code}]')
        try:
            query_result = response.json()
            if not query_result['success'] or len(query_result['Result']["Results"]) == 0:
                raise ArkServiceException('Failed to retrieve user id by name')
            user_row = query_result['Result']["Results"][0]["Row"]
            last_login = None
            if last_login := user_row.get('LastLogin'):
                try:
                    last_login = last_login.split('(')[1].split(')')[0]
                    last_login = f'{last_login[:10]}.{last_login[10:]}'  # for milliseconds
                    last_login = datetime.fromtimestamp(float(last_login), timezone.utc)
                except Exception as ex:
                    self._logger.debug(f'Failed to parse last login [{user_row.get("LastLogin")}] [{str(ex)}]')

            return ArkIdentityUser(
                user_id=user_row["ID"],
                username=user_row["Username"],
                display_name=user_row["DisplayName"],
                email=user_row["Email"],
                mobile_number=user_row["MobileNumber"],
                last_login=last_login,
            )
        except (ValidationError, JSONDecodeError, KeyError) as ex:
            self._logger.exception(f'Failed to parse user id by name response [{str(ex)}] - [{response.text}]')
            raise ArkServiceException(f'Failed to parse user id by name response [{str(ex)}]') from ex

    def user_by_id(self, user_by_id: ArkIdentityUserById) -> ArkIdentityUser:
        """
        Finds the identifier of the given id

        Args:
            user_by_id (ArkIdentityUserIdById): _description_

        Returns:
            str: _description_
        """
        response: Response = self._client.post(
            f'{self._url_prefix}{REDROCK_QUERY}',
            json={"Script": f"Select ID, Username, DisplayName, Email, MobileNumber, LastLogin from User WHERE ID='{user_by_id.user_id}'"},
        )
        if response.status_code != HTTPStatus.OK:
            raise ArkServiceException(f'Failed to retrieve user id by id [{response.text}] - [{response.status_code}]')
        try:
            query_result = response.json()
            if not query_result['success'] or len(query_result['Result']["Results"]) == 0:
                raise ArkServiceException('Failed to retrieve user id by id')
            user_row = query_result['Result']["Results"][0]["Row"]
            last_login = None
            if last_login := user_row.get('LastLogin'):
                try:
                    last_login = last_login.split('(')[1].split(')')[0]
                    last_login = f'{last_login[:10]}.{last_login[10:]}'  # for milliseconds
                    last_login = datetime.fromtimestamp(float(last_login), timezone.utc)
                except Exception as ex:
                    self._logger.debug(f'Failed to parse last login [{user_row.get("LastLogin")}] [{str(ex)}]')

            return ArkIdentityUser(
                user_id=user_row["ID"],
                username=user_row["Username"],
                display_name=user_row["DisplayName"],
                email=user_row["Email"],
                mobile_number=user_row["MobileNumber"],
                last_login=last_login,
            )
        except (ValidationError, JSONDecodeError, KeyError) as ex:
            self._logger.exception(f'Failed to parse user id by id response [{str(ex)}] - [{response.text}]')
            raise ArkServiceException(f'Failed to parse user id by id response [{str(ex)}]') from ex

    def reset_user_password(self, reset_user_password: ArkIdentityResetUserPassword) -> None:
        """
        Resets a given username's password to the new given one
        Assumes the logged in user has permissions to do so

        Args:
            reset_user_password (ArkIdentityResetUserPassword): _description_

        Raises:
            ArkServiceException: _description_
        """
        user_id = self.user_id_by_name(ArkIdentityUserIdByName(username=reset_user_password.username))
        response: Response = self._client.post(
            f'{self._url_prefix}{RESET_USER_PASSWORD_URL}', json={'ID': user_id, 'newPassword': reset_user_password.new_password}
        )
        try:
            result = response.json()
            if response.status_code != HTTPStatus.OK or not result['success']:
                raise ArkServiceException(f'Failed to reset user password [{response.text}]')
        except (ValidationError, JSONDecodeError, KeyError) as ex:
            self._logger.exception(f'Failed to parse reset user password response [{str(ex)}] - [{response.text}]')
            raise ArkServiceException(f'Failed to parse reset user password response [{str(ex)}]') from ex

    def user_info(self) -> ArkIdentityUserInfo:
        """
        Retrieves the current user info

        Raises:
            ArkServiceException: _description_

        Returns:
            ArkIdentityUserInfo: _description_
        """
        response: Response = self._client.post(
            f'{self._url_prefix}{USER_INFO_URL}',
            json={'Scopes': ['userInfo']},
        )
        try:
            result = response.json()
            if response.status_code != HTTPStatus.OK:
                raise ArkServiceException(f'Failed to get user info [{response.text}]')
            return ArkIdentityUserInfo.model_validate(result)
        except (ValidationError, JSONDecodeError, KeyError) as ex:
            self._logger.exception(f'Failed to get user info [{str(ex)}] - [{response.text}]')
            raise ArkServiceException(f'Failed to get user info [{str(ex)}]') from ex

    @staticmethod
    @overrides
    def service_config() -> ArkServiceConfig:
        return SERVICE_CONFIG

create_user(create_user)

Creates a user with the given details, and returns its finalized details and id

Parameters:

Name Type Description Default
create_user ArkIdentityCreateUser

description

 required

Raises:

Type Description
ArkServiceException

description

Returns:

Name Type Description
ArkIdentityUser ArkIdentityUser

description
Source code in ark_sdk_python/services/identity/users/ark_identity_users_service.py 
44
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
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
def create_user(self, create_user: ArkIdentityCreateUser) -> ArkIdentityUser:
    """
    Creates a user with the given details, and returns its finalized details and id

    Args:
        create_user (ArkIdentityCreateUser): _description_

    Raises:
        ArkServiceException: _description_

    Returns:
        ArkIdentityUser: _description_
    """
    self._logger.info(f'Creating identity user [{create_user.username}]')
    directories_service = ArkIdentityDirectoriesService(self._isp_auth)
    tenant_suffix = create_user.suffix or directories_service.tenant_default_suffix()
    response: Response = self._client.post(
        f'{self._url_prefix}{CREATE_USER_URL}',
        json={
            "DisplayName": create_user.display_name,
            "Name": f'{create_user.username}@{tenant_suffix}',
            "Mail": create_user.email,
            "Password": create_user.password.get_secret_value(),
            "MobileNumber": create_user.mobile_number,
            "InEverybodyRole": 'true',
            "InSysAdminRole": 'false',
            "ForcePasswordChangeNext": 'false',
            "SendEmailInvite": 'false',
            "SendSmsInvite": 'false',
        },
    )
    try:
        result = response.json()
        if response.status_code != HTTPStatus.OK or not result['success']:
            raise ArkServiceException(f'Failed to create user [{response.text}]')
        if create_user.roles:
            roles_service = ArkIdentityRolesService(self._isp_auth)
            for role in create_user.roles:
                roles_service.add_user_to_role(
                    ArkIdentityAddUserToRole(username=f'{create_user.username}@{tenant_suffix}', role_name=role)
                )
        self._logger.info(f'User created successfully with id [{result["Result"]}]')
        return ArkIdentityUser(
            user_id=result['Result'],
            username=f'{create_user.username}@{tenant_suffix}',
            display_name=create_user.display_name,
            email=create_user.email,
            mobile_number=create_user.mobile_number,
            roles=create_user.roles,
        )
    except (ValidationError, JSONDecodeError, KeyError) as ex:
        self._logger.exception(f'Failed to parse create user response [{str(ex)}] - [{response.text}]')
        raise ArkServiceException(f'Failed to parse create user response [{str(ex)}]') from ex

delete_user(delete_user)

Deletes a user by given name

Parameters:

Name Type Description Default
delete_user ArkIdentityDeleteUser

description

 required

Raises:

Type Description
ArkServiceException

description
Source code in ark_sdk_python/services/identity/users/ark_identity_users_service.py 
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
def delete_user(self, delete_user: ArkIdentityDeleteUser) -> None:
    """
    Deletes a user by given name

    Args:
        delete_user (ArkIdentityDeleteUser): _description_

    Raises:
        ArkServiceException: _description_
    """
    self._logger.info(f'Deleting user [{delete_user.user_id}]')
    response: Response = self._client.post(
        f'{self._url_prefix}{DELETE_USER_URL}', json={'ID': delete_user.user_id or delete_user.username}
    )
    try:
        if response.status_code != HTTPStatus.OK or not response.json()['success']:
            raise ArkServiceException(f'Failed to delete user [{response.text}]')
    except (ValidationError, JSONDecodeError, KeyError) as ex:
        self._logger.exception(f'Failed to parse delete user response [{str(ex)}] - [{response.text}]')
        raise ArkServiceException(f'Failed to parse delete user response [{str(ex)}]') from ex

delete_users(delete_users)

Deletes users by given ids

Parameters:

Name Type Description Default
delete_users ArkIdentityDeleteUsers

description

 required

Raises:

Type Description
ArkServiceException

description
Source code in ark_sdk_python/services/identity/users/ark_identity_users_service.py 
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
def delete_users(self, delete_users: ArkIdentityDeleteUsers) -> None:
    """
    Deletes users by given ids

    Args:
        delete_users (ArkIdentityDeleteUsers): _description_

    Raises:
        ArkServiceException: _description_
    """
    self._logger.info(f'Removing users [{",".join(delete_users.user_ids)}]')
    response: Response = self._client.post(f'{self._url_prefix}{REMOVE_USERS_URL}', json={'Users': [delete_users.user_ids]})
    try:
        if response.status_code != HTTPStatus.OK or not response.json()['success']:
            raise ArkServiceException(f'Failed to remove users [{response.text}]')
    except (ValidationError, JSONDecodeError, KeyError) as ex:
        self._logger.exception(f'Failed to parse remove users response [{str(ex)}] - [{response.text}]')
        raise ArkServiceException(f'Failed to parse remove users response [{str(ex)}]') from ex

reset_user_password(reset_user_password)

Resets a given username's password to the new given one Assumes the logged in user has permissions to do so

Parameters:

Name Type Description Default
reset_user_password ArkIdentityResetUserPassword

description

 required

Raises:

Type Description
ArkServiceException

description
Source code in ark_sdk_python/services/identity/users/ark_identity_users_service.py 
285
286
287
288
289
290
291
292
293
294
295
296
297
298
299
300
301
302
303
304
305
306
def reset_user_password(self, reset_user_password: ArkIdentityResetUserPassword) -> None:
    """
    Resets a given username's password to the new given one
    Assumes the logged in user has permissions to do so

    Args:
        reset_user_password (ArkIdentityResetUserPassword): _description_

    Raises:
        ArkServiceException: _description_
    """
    user_id = self.user_id_by_name(ArkIdentityUserIdByName(username=reset_user_password.username))
    response: Response = self._client.post(
        f'{self._url_prefix}{RESET_USER_PASSWORD_URL}', json={'ID': user_id, 'newPassword': reset_user_password.new_password}
    )
    try:
        result = response.json()
        if response.status_code != HTTPStatus.OK or not result['success']:
            raise ArkServiceException(f'Failed to reset user password [{response.text}]')
    except (ValidationError, JSONDecodeError, KeyError) as ex:
        self._logger.exception(f'Failed to parse reset user password response [{str(ex)}] - [{response.text}]')
        raise ArkServiceException(f'Failed to parse reset user password response [{str(ex)}]') from ex

update_user(update_user)

Updates the user information

Parameters:

Name Type Description Default
update_user ArkIdentityUpdateUser

description

 required

Raises:

Type Description
ArkServiceException

description
Source code in ark_sdk_python/services/identity/users/ark_identity_users_service.py 
 98
 99
100
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
129
130
131
132
def update_user(self, update_user: ArkIdentityUpdateUser) -> None:
    """
    Updates the user information

    Args:
        update_user (ArkIdentityUpdateUser): _description_

    Raises:
        ArkServiceException: _description_
    """
    if update_user.username and not update_user.user_id:
        update_user.user_id = self.user_id_by_name(ArkIdentityUserIdByName(username=update_user.username))
    self._logger.info(f'Updating identity user [{update_user.user_id}]')
    update_dict = {}
    if update_user.new_username:
        if '@' not in update_user.new_username:
            tenant_suffix = update_user.username.split('@')[1]
            update_user.new_username = f'{update_user.new_username}@{tenant_suffix}'
        update_dict['Name'] = update_user.new_username
    if update_user.display_name:
        update_dict['DisplayName'] = update_user.display_name
    if update_user.email:
        update_dict['Mail'] = update_user.email
    if update_user.mobile_number:
        update_dict['MobileNumber'] = update_user.mobile_number
    update_dict['ID'] = update_user.user_id
    response: Response = self._client.post(f'{self._url_prefix}{UPDATE_USER_URL}', json=update_dict)
    try:
        result = response.json()
        if response.status_code != HTTPStatus.OK or not result['success']:
            raise ArkServiceException(f'Failed to update user [{response.text}]')
        self._logger.info('User updated successfully')
    except (ValidationError, JSONDecodeError, KeyError) as ex:
        self._logger.exception(f'Failed to parse update user response [{str(ex)}] - [{response.text}]')
        raise ArkServiceException(f'Failed to parse update user response [{str(ex)}]') from ex

user_by_id(user_by_id)

Finds the identifier of the given id

Parameters:

Name Type Description Default
user_by_id ArkIdentityUserIdById

description

 required

Returns:

Name Type Description
str ArkIdentityUser

description
Source code in ark_sdk_python/services/identity/users/ark_identity_users_service.py 
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
def user_by_id(self, user_by_id: ArkIdentityUserById) -> ArkIdentityUser:
    """
    Finds the identifier of the given id

    Args:
        user_by_id (ArkIdentityUserIdById): _description_

    Returns:
        str: _description_
    """
    response: Response = self._client.post(
        f'{self._url_prefix}{REDROCK_QUERY}',
        json={"Script": f"Select ID, Username, DisplayName, Email, MobileNumber, LastLogin from User WHERE ID='{user_by_id.user_id}'"},
    )
    if response.status_code != HTTPStatus.OK:
        raise ArkServiceException(f'Failed to retrieve user id by id [{response.text}] - [{response.status_code}]')
    try:
        query_result = response.json()
        if not query_result['success'] or len(query_result['Result']["Results"]) == 0:
            raise ArkServiceException('Failed to retrieve user id by id')
        user_row = query_result['Result']["Results"][0]["Row"]
        last_login = None
        if last_login := user_row.get('LastLogin'):
            try:
                last_login = last_login.split('(')[1].split(')')[0]
                last_login = f'{last_login[:10]}.{last_login[10:]}'  # for milliseconds
                last_login = datetime.fromtimestamp(float(last_login), timezone.utc)
            except Exception as ex:
                self._logger.debug(f'Failed to parse last login [{user_row.get("LastLogin")}] [{str(ex)}]')

        return ArkIdentityUser(
            user_id=user_row["ID"],
            username=user_row["Username"],
            display_name=user_row["DisplayName"],
            email=user_row["Email"],
            mobile_number=user_row["MobileNumber"],
            last_login=last_login,
        )
    except (ValidationError, JSONDecodeError, KeyError) as ex:
        self._logger.exception(f'Failed to parse user id by id response [{str(ex)}] - [{response.text}]')
        raise ArkServiceException(f'Failed to parse user id by id response [{str(ex)}]') from ex

user_by_name(user_id_by_name)

Finds the identifier of the given username

Parameters:

Name Type Description Default
user_id_by_name ArkIdentityUserIdByName

description

 required

Returns:

Name Type Description
str ArkIdentityUser

description
Source code in ark_sdk_python/services/identity/users/ark_identity_users_service.py 
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
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
def user_by_name(self, user_id_by_name: ArkIdentityUserByName) -> ArkIdentityUser:
    """
    Finds the identifier of the given username

    Args:
        user_id_by_name (ArkIdentityUserIdByName): _description_

    Returns:
        str: _description_
    """
    response: Response = self._client.post(
        f'{self._url_prefix}{REDROCK_QUERY}',
        json={
            "Script": f"Select ID, Username, DisplayName, Email, MobileNumber, LastLogin from User WHERE Username='{user_id_by_name.username}'"
        },
    )
    if response.status_code != HTTPStatus.OK:
        raise ArkServiceException(f'Failed to retrieve user id by name [{response.text}] - [{response.status_code}]')
    try:
        query_result = response.json()
        if not query_result['success'] or len(query_result['Result']["Results"]) == 0:
            raise ArkServiceException('Failed to retrieve user id by name')
        user_row = query_result['Result']["Results"][0]["Row"]
        last_login = None
        if last_login := user_row.get('LastLogin'):
            try:
                last_login = last_login.split('(')[1].split(')')[0]
                last_login = f'{last_login[:10]}.{last_login[10:]}'  # for milliseconds
                last_login = datetime.fromtimestamp(float(last_login), timezone.utc)
            except Exception as ex:
                self._logger.debug(f'Failed to parse last login [{user_row.get("LastLogin")}] [{str(ex)}]')

        return ArkIdentityUser(
            user_id=user_row["ID"],
            username=user_row["Username"],
            display_name=user_row["DisplayName"],
            email=user_row["Email"],
            mobile_number=user_row["MobileNumber"],
            last_login=last_login,
        )
    except (ValidationError, JSONDecodeError, KeyError) as ex:
        self._logger.exception(f'Failed to parse user id by name response [{str(ex)}] - [{response.text}]')
        raise ArkServiceException(f'Failed to parse user id by name response [{str(ex)}]') from ex

user_id_by_name(user_id_by_name)

Finds the identifier of the given username

Parameters:

Name Type Description Default
user_id_by_name ArkIdentityUserIdByName

description

 required

Returns:

Name Type Description
str str

description
Source code in ark_sdk_python/services/identity/users/ark_identity_users_service.py 
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
def user_id_by_name(self, user_id_by_name: ArkIdentityUserIdByName) -> str:
    """
    Finds the identifier of the given username

    Args:
        user_id_by_name (ArkIdentityUserIdByName): _description_

    Returns:
        str: _description_
    """
    response: Response = self._client.post(
        f'{self._url_prefix}{REDROCK_QUERY}',
        json={"Script": f"Select ID, Username from User WHERE Username='{user_id_by_name.username}'"},
    )
    if response.status_code != HTTPStatus.OK:
        raise ArkServiceException(f'Failed to retrieve user id by name [{response.text}] - [{response.status_code}]')
    try:
        query_result = response.json()
        if not query_result['success'] or len(query_result['Result']["Results"]) == 0:
            raise ArkServiceException('Failed to retrieve user id by name')
        return query_result['Result']["Results"][0]["Row"]["ID"]
    except (ValidationError, JSONDecodeError, KeyError) as ex:
        self._logger.exception(f'Failed to parse user id by name response [{str(ex)}] - [{response.text}]')
        raise ArkServiceException(f'Failed to parse user id by name response [{str(ex)}]') from ex

user_info()

Retrieves the current user info

Raises:

Type Description
ArkServiceException

description

Returns:

Name Type Description
ArkIdentityUserInfo ArkIdentityUserInfo

description
Source code in ark_sdk_python/services/identity/users/ark_identity_users_service.py 
308
309
310
311
312
313
314
315
316
317
318
319
320
321
322
323
324
325
326
327
328
329
def user_info(self) -> ArkIdentityUserInfo:
    """
    Retrieves the current user info

    Raises:
        ArkServiceException: _description_

    Returns:
        ArkIdentityUserInfo: _description_
    """
    response: Response = self._client.post(
        f'{self._url_prefix}{USER_INFO_URL}',
        json={'Scopes': ['userInfo']},
    )
    try:
        result = response.json()
        if response.status_code != HTTPStatus.OK:
            raise ArkServiceException(f'Failed to get user info [{response.text}]')
        return ArkIdentityUserInfo.model_validate(result)
    except (ValidationError, JSONDecodeError, KeyError) as ex:
        self._logger.exception(f'Failed to get user info [{str(ex)}] - [{response.text}]')
        raise ArkServiceException(f'Failed to get user info [{str(ex)}]') from ex