Skip to content

ark_identity_users_service

ArkIdentityUsersService

Bases: ArkIdentityBaseService

Source code in ark_sdk_python/services/identity/users/ark_identity_users_service.py 
 40
 41
 42
 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
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_
        """
        if delete_user.username and not delete_user.user_id:
            delete_user.user_id = self.user_id_by_name(ArkIdentityUserIdByName(username=delete_user.username))
        self._logger.info(f'Deleting user [{delete_user.user_id}]')
        response: Response = self._client.post(f'{self._url_prefix}{DELETE_USER_URL}', json={'Users': [delete_user.user_id]})
        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 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 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_
        """
        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.parse_obj(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 
41
42
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
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 
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
def delete_user(self, delete_user: ArkIdentityDeleteUser) -> None:
    """
    Deletes a user by given name

    Args:
        delete_user (ArkIdentityDeleteUser): _description_

    Raises:
        ArkServiceException: _description_
    """
    if delete_user.username and not delete_user.user_id:
        delete_user.user_id = self.user_id_by_name(ArkIdentityUserIdByName(username=delete_user.username))
    self._logger.info(f'Deleting user [{delete_user.user_id}]')
    response: Response = self._client.post(f'{self._url_prefix}{DELETE_USER_URL}', json={'Users': [delete_user.user_id]})
    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

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 
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
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 
 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
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_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 
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
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 
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
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
Source code in ark_sdk_python/services/identity/users/ark_identity_users_service.py 
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
def user_info(self) -> ArkIdentityUserInfo:
    """
    Retrieves the current user info

    Raises:
        ArkServiceException: _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.parse_obj(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