Skip to content

ark_dpa_db_workspace_service

ArkDPADBWorkspaceService

Bases: ArkService

Source code in ark_sdk_python/services/dpa/workspaces/db/ark_dpa_db_workspace_service.py 
 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
class ArkDPADBWorkspaceService(ArkService):
    def __init__(self, isp_auth: ArkISPAuth) -> None:
        super().__init__(isp_auth)
        self.__isp_auth = isp_auth
        self.__client: ArkISPServiceClient = ArkISPServiceClient.from_isp_auth(self.__isp_auth, 'dpa')

    def __list_databases_with_filters(
        self, provider_family: Optional[ArkDPADBDatabaseFamilyType] = None, tags: Optional[List[ArkDPADBTag]] = None
    ) -> ArkDPADBDatabaseInfoList:
        params = {}
        if provider_family:
            params['provider-family'] = provider_family.value
        if tags:
            for tag in tags:
                params[f'key.{tag.key}'] = tag.value
        resp: Response = self.__client.get(RESOURCES_API, params=params)
        if resp.status_code == HTTPStatus.OK:
            try:
                return ArkDPADBDatabaseInfoList.parse_obj(resp.json())
            except (ValidationError, JSONDecodeError, KeyError) as ex:
                self._logger.exception(f'Failed to parse list databases response [{str(ex)}] - [{resp.text}]')
                raise ArkServiceException(f'Failed to parse list databases response [{str(ex)}]') from ex
        raise ArkServiceException(f'Failed to list databases [{resp.text}] - [{resp.status_code}]')

    def add_database(self, add_database: ArkDPADBAddDatabase) -> ArkDPADBDatabase:
        """
        Adds a new database.

        Args:
            add_database (ArkDPADBAddDatabase): _description_

        Raises:
            ArkServiceException: _description_

        Returns:
            ArkDPADBDatabase: _description_
        """
        self._logger.info(f'Adding database [{add_database.name}]')
        if not add_database.port:
            add_database.port = DATABASE_FAMILIES_DEFAULT_PORTS[DATABASES_ENGINES_TO_FAMILY[add_database.provider_engine]]
        add_database_dict = add_database.dict(exclude_none=True)
        add_database_dict['platform'] = serialize_db_platform_type(add_database.platform)
        resp: Response = self.__client.post(RESOURCES_API, json=add_database_dict)
        if resp.status_code == HTTPStatus.CREATED:
            try:
                database_id = resp.json()['target_id']
                return self.database(ArkDPADBGetDatabase(id=database_id))
            except (ValidationError, JSONDecodeError, KeyError) as ex:
                self._logger.exception(f'Failed to parse add database response [{str(ex)}] - [{resp.text}]')
                raise ArkServiceException(f'Failed to parse add database response [{str(ex)}]') from ex
        raise ArkServiceException(f'Failed to add database [{resp.text}] - [{resp.status_code}]')

    def delete_database(self, delete_database: ArkDPADBDeleteDatabase) -> None:
        """
        Deletes a database.

        Args:
            delete_database (ArkDPADBDeleteDatabase): _description_

        Raises:
            ArkServiceException: _description_
        """
        if delete_database.name and not delete_database.id:
            delete_database.id = self.list_databases_by(ArkDPADBDatabasesFilter(name=delete_database.name))
        self._logger.info(f'Deleting database [{delete_database.id}]')
        resp: Response = self.__client.delete(RESOURCE_API.format(resource_id=delete_database.id))
        if resp.status_code != HTTPStatus.NO_CONTENT:
            raise ArkServiceException(f'Failed to delete database [{resp.text}] - [{resp.status_code}]')

    def update_database(self, update_database: ArkDPADBUpdateDatabase) -> ArkDPADBDatabase:
        """
        Updates a database.

        Args:
            update_database (ArkDPADBUpdateDatabase): _description_

        Raises:
            ArkServiceException: _description_

        Returns:
            ArkDPADBDatabase: _description_
        """
        if update_database.name and not update_database.id:
            databases = self.list_databases_by(ArkDPADBDatabasesFilter(name=update_database.name))
            if not databases.items or len(databases.items) != 1:
                raise ArkServiceException(f'Failed to update database - name [{update_database.name}] not found')
            update_database.id = self.list_databases_by(ArkDPADBDatabasesFilter(name=update_database.name)).items[0].id
        self._logger.info(f'Updating database [{update_database.id}]')
        update_database_dict = update_database.dict(exclude={'name', 'new_name'}, exclude_none=True)
        if update_database.new_name:
            update_database_dict["name"] = update_database.new_name
        resp: Response = self.__client.put(RESOURCE_API.format(resource_id=update_database.id), json=update_database_dict)
        if resp.status_code == HTTPStatus.OK:
            try:
                return self.database(ArkDPADBGetDatabase(id=update_database.id))
            except (ValidationError, JSONDecodeError, KeyError) as ex:
                self._logger.exception(f'Failed to parse update database response [{str(ex)}] - [{resp.text}]')
                raise ArkServiceException(f'Failed to parse update database response [{str(ex)}]') from ex
        raise ArkServiceException(f'Failed to update database [{resp.text}] - [{resp.status_code}]')

    def list_databases(self) -> ArkDPADBDatabaseInfoList:
        """
        Lists all databases.

        Raises:
            ArkServiceException: _description_

        Returns:
            ArkDPADBDatabaseInfoList: _description_
        """
        self._logger.info('Listing all databases')
        return self.__list_databases_with_filters()

    def list_databases_by(self, databases_filter: ArkDPADBDatabasesFilter) -> ArkDPADBDatabaseInfoList:
        """
        Lists databases that match the specified filters.

        Args:
            databases_filter (ArkDPADBDatabasesFilter): _description_

        Returns:
            ArkDPADBDatabaseInfoList: _description_
        """
        self._logger.info(f'Listing databases by filters [{databases_filter}]')
        databases = self.__list_databases_with_filters(databases_filter.provider_family, databases_filter.tags)
        if databases_filter.name:
            databases.items = [d for d in databases.items if fnmatch(d.name, databases_filter.name)]
        if databases_filter.provider_engine:
            databases.items = [d for d in databases.items if d.provider_info.engine == databases_filter.provider_engine]
        if databases_filter.provider_workspace:
            databases.items = [d for d in databases.items if d.provider_info.workspace == databases_filter.provider_workspace]
        if databases_filter.auth_methods:
            databases.items = [d for d in databases.items if d.configured_auth_method_type in databases_filter.auth_methods]
        if databases_filter.db_warnings_filter:
            if databases_filter.db_warnings_filter in (
                ArkDPADBWarning.AnyError,
                ArkDPADBWarning.NoCertificates,
            ):
                databases.items = [d for d in databases.items if not d.certificate]
            if databases_filter.db_warnings_filter in (
                ArkDPADBWarning.AnyError,
                ArkDPADBWarning.NoSecrets,
            ):
                databases.items = [d for d in databases.items if not d.secret_id]
        databases.total_count = len(databases.items)
        return databases

    def database(self, get_database: ArkDPADBGetDatabase) -> ArkDPADBDatabase:
        """
        Gets a specific database.

        Args:
            get_database (ArkDPADBGetDatabase): _description_

        Raises:
            ArkServiceException: _description_

        Returns:
            ArkDPADBDatabase: _description_
        """
        if get_database.name and not get_database.id:
            databases = self.list_databases_by(ArkDPADBDatabasesFilter(name=get_database.name))
            if not databases.items or len(databases.items) != 1:
                raise ArkServiceException(f'Failed to get database - name [{get_database.name}] not found')
            get_database.id = self.list_databases_by(ArkDPADBDatabasesFilter(name=get_database.name)).items[0].id
        self._logger.info(f'Getting database [{get_database.id}]')
        resp: Response = self.__client.get(RESOURCE_API.format(resource_id=get_database.id))
        if resp.status_code == HTTPStatus.OK:
            try:
                return ArkDPADBDatabase.parse_obj(resp.json())
            except (ValidationError, JSONDecodeError, KeyError) as ex:
                self._logger.exception(f'Failed to parse database response [{str(ex)}] - [{resp.text}]')
                raise ArkServiceException(f'Failed to parse database response [{str(ex)}]') from ex
        raise ArkServiceException(f'Failed to get database [{resp.text}] - [{resp.status_code}]')

    def databases_stats(self) -> ArkDPADBDatabasesStats:
        """
        Calculate statistics about the tenant's databases.

        Returns:
            ArkDPADBDatabasesStats: _description_
        """
        self._logger.info('Calculating databases stats')
        databases = self.list_databases()
        databases_stats = ArkDPADBDatabasesStats.construct()
        databases_stats.databases_count = len(databases.items)

        # Get databases per engine
        engines_types: Set[ArkDPADBDatabaseEngineType] = {d.provider_info.engine for d in databases.items}
        databases_stats.databases_count_by_engine = {
            et: len([d for d in databases.items if d.provider_info.engine == et]) for et in engines_types
        }

        # Get databases per workspace
        workspaces_types: Set[ArkDPADBDatabaseWorkspaceType] = {d.provider_info.workspace for d in databases.items}
        databases_stats.databases_count_by_workspace = {
            wt: len([d for d in databases.items if d.provider_info.workspace == wt]) for wt in workspaces_types
        }

        # Get databases per family
        family_types: Set[ArkDPADBDatabaseFamilyType] = {d.provider_info.family for d in databases.items}
        databases_stats.databases_count_by_family = {
            ft: len([d for d in databases.items if d.provider_info.family == ft]) for ft in family_types
        }

        # Get databases per auth method
        auth_method_types: Set[ArkDPADBAuthMethodType] = {d.configured_auth_method_type for d in databases.items}
        databases_stats.databases_count_by_auth_method = {
            am: len([d for d in databases.items if d.configured_auth_method_type == am]) for am in auth_method_types
        }

        # Get databases per db warning
        databases_stats.databases_count_by_warning = {
            ArkDPADBWarning.NoCertificates: len([d for d in databases.items if not d.certificate]),
            ArkDPADBWarning.NoSecrets: len([d for d in databases.items if not d.secret_id]),
        }

        return databases_stats

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

add_database(add_database)

Adds a new database.

Parameters:

Name Type Description Default
add_database ArkDPADBAddDatabase

description

 required

Raises:

Type Description
ArkServiceException

description

Returns:

Name Type Description
ArkDPADBDatabase ArkDPADBDatabase

description
Source code in ark_sdk_python/services/dpa/workspaces/db/ark_dpa_db_workspace_service.py 
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
def add_database(self, add_database: ArkDPADBAddDatabase) -> ArkDPADBDatabase:
    """
    Adds a new database.

    Args:
        add_database (ArkDPADBAddDatabase): _description_

    Raises:
        ArkServiceException: _description_

    Returns:
        ArkDPADBDatabase: _description_
    """
    self._logger.info(f'Adding database [{add_database.name}]')
    if not add_database.port:
        add_database.port = DATABASE_FAMILIES_DEFAULT_PORTS[DATABASES_ENGINES_TO_FAMILY[add_database.provider_engine]]
    add_database_dict = add_database.dict(exclude_none=True)
    add_database_dict['platform'] = serialize_db_platform_type(add_database.platform)
    resp: Response = self.__client.post(RESOURCES_API, json=add_database_dict)
    if resp.status_code == HTTPStatus.CREATED:
        try:
            database_id = resp.json()['target_id']
            return self.database(ArkDPADBGetDatabase(id=database_id))
        except (ValidationError, JSONDecodeError, KeyError) as ex:
            self._logger.exception(f'Failed to parse add database response [{str(ex)}] - [{resp.text}]')
            raise ArkServiceException(f'Failed to parse add database response [{str(ex)}]') from ex
    raise ArkServiceException(f'Failed to add database [{resp.text}] - [{resp.status_code}]')

database(get_database)

Gets a specific database.

Parameters:

Name Type Description Default
get_database ArkDPADBGetDatabase

description

 required

Raises:

Type Description
ArkServiceException

description

Returns:

Name Type Description
ArkDPADBDatabase ArkDPADBDatabase

description
Source code in ark_sdk_python/services/dpa/workspaces/db/ark_dpa_db_workspace_service.py 
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
def database(self, get_database: ArkDPADBGetDatabase) -> ArkDPADBDatabase:
    """
    Gets a specific database.

    Args:
        get_database (ArkDPADBGetDatabase): _description_

    Raises:
        ArkServiceException: _description_

    Returns:
        ArkDPADBDatabase: _description_
    """
    if get_database.name and not get_database.id:
        databases = self.list_databases_by(ArkDPADBDatabasesFilter(name=get_database.name))
        if not databases.items or len(databases.items) != 1:
            raise ArkServiceException(f'Failed to get database - name [{get_database.name}] not found')
        get_database.id = self.list_databases_by(ArkDPADBDatabasesFilter(name=get_database.name)).items[0].id
    self._logger.info(f'Getting database [{get_database.id}]')
    resp: Response = self.__client.get(RESOURCE_API.format(resource_id=get_database.id))
    if resp.status_code == HTTPStatus.OK:
        try:
            return ArkDPADBDatabase.parse_obj(resp.json())
        except (ValidationError, JSONDecodeError, KeyError) as ex:
            self._logger.exception(f'Failed to parse database response [{str(ex)}] - [{resp.text}]')
            raise ArkServiceException(f'Failed to parse database response [{str(ex)}]') from ex
    raise ArkServiceException(f'Failed to get database [{resp.text}] - [{resp.status_code}]')

databases_stats()

Calculate statistics about the tenant's databases.

Returns:

Name Type Description
ArkDPADBDatabasesStats ArkDPADBDatabasesStats

description
Source code in ark_sdk_python/services/dpa/workspaces/db/ark_dpa_db_workspace_service.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
256
257
258
259
def databases_stats(self) -> ArkDPADBDatabasesStats:
    """
    Calculate statistics about the tenant's databases.

    Returns:
        ArkDPADBDatabasesStats: _description_
    """
    self._logger.info('Calculating databases stats')
    databases = self.list_databases()
    databases_stats = ArkDPADBDatabasesStats.construct()
    databases_stats.databases_count = len(databases.items)

    # Get databases per engine
    engines_types: Set[ArkDPADBDatabaseEngineType] = {d.provider_info.engine for d in databases.items}
    databases_stats.databases_count_by_engine = {
        et: len([d for d in databases.items if d.provider_info.engine == et]) for et in engines_types
    }

    # Get databases per workspace
    workspaces_types: Set[ArkDPADBDatabaseWorkspaceType] = {d.provider_info.workspace for d in databases.items}
    databases_stats.databases_count_by_workspace = {
        wt: len([d for d in databases.items if d.provider_info.workspace == wt]) for wt in workspaces_types
    }

    # Get databases per family
    family_types: Set[ArkDPADBDatabaseFamilyType] = {d.provider_info.family for d in databases.items}
    databases_stats.databases_count_by_family = {
        ft: len([d for d in databases.items if d.provider_info.family == ft]) for ft in family_types
    }

    # Get databases per auth method
    auth_method_types: Set[ArkDPADBAuthMethodType] = {d.configured_auth_method_type for d in databases.items}
    databases_stats.databases_count_by_auth_method = {
        am: len([d for d in databases.items if d.configured_auth_method_type == am]) for am in auth_method_types
    }

    # Get databases per db warning
    databases_stats.databases_count_by_warning = {
        ArkDPADBWarning.NoCertificates: len([d for d in databases.items if not d.certificate]),
        ArkDPADBWarning.NoSecrets: len([d for d in databases.items if not d.secret_id]),
    }

    return databases_stats

delete_database(delete_database)

Deletes a database.

Parameters:

Name Type Description Default
delete_database ArkDPADBDeleteDatabase

description

 required

Raises:

Type Description
ArkServiceException

description
Source code in ark_sdk_python/services/dpa/workspaces/db/ark_dpa_db_workspace_service.py 
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
def delete_database(self, delete_database: ArkDPADBDeleteDatabase) -> None:
    """
    Deletes a database.

    Args:
        delete_database (ArkDPADBDeleteDatabase): _description_

    Raises:
        ArkServiceException: _description_
    """
    if delete_database.name and not delete_database.id:
        delete_database.id = self.list_databases_by(ArkDPADBDatabasesFilter(name=delete_database.name))
    self._logger.info(f'Deleting database [{delete_database.id}]')
    resp: Response = self.__client.delete(RESOURCE_API.format(resource_id=delete_database.id))
    if resp.status_code != HTTPStatus.NO_CONTENT:
        raise ArkServiceException(f'Failed to delete database [{resp.text}] - [{resp.status_code}]')

list_databases()

Lists all databases.

Raises:

Type Description
ArkServiceException

description

Returns:

Name Type Description
ArkDPADBDatabaseInfoList ArkDPADBDatabaseInfoList

description
Source code in ark_sdk_python/services/dpa/workspaces/db/ark_dpa_db_workspace_service.py 
142
143
144
145
146
147
148
149
150
151
152
153
def list_databases(self) -> ArkDPADBDatabaseInfoList:
    """
    Lists all databases.

    Raises:
        ArkServiceException: _description_

    Returns:
        ArkDPADBDatabaseInfoList: _description_
    """
    self._logger.info('Listing all databases')
    return self.__list_databases_with_filters()

list_databases_by(databases_filter)

Lists databases that match the specified filters.

Parameters:

Name Type Description Default
databases_filter ArkDPADBDatabasesFilter

description

 required

Returns:

Name Type Description
ArkDPADBDatabaseInfoList ArkDPADBDatabaseInfoList

description
Source code in ark_sdk_python/services/dpa/workspaces/db/ark_dpa_db_workspace_service.py 
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
def list_databases_by(self, databases_filter: ArkDPADBDatabasesFilter) -> ArkDPADBDatabaseInfoList:
    """
    Lists databases that match the specified filters.

    Args:
        databases_filter (ArkDPADBDatabasesFilter): _description_

    Returns:
        ArkDPADBDatabaseInfoList: _description_
    """
    self._logger.info(f'Listing databases by filters [{databases_filter}]')
    databases = self.__list_databases_with_filters(databases_filter.provider_family, databases_filter.tags)
    if databases_filter.name:
        databases.items = [d for d in databases.items if fnmatch(d.name, databases_filter.name)]
    if databases_filter.provider_engine:
        databases.items = [d for d in databases.items if d.provider_info.engine == databases_filter.provider_engine]
    if databases_filter.provider_workspace:
        databases.items = [d for d in databases.items if d.provider_info.workspace == databases_filter.provider_workspace]
    if databases_filter.auth_methods:
        databases.items = [d for d in databases.items if d.configured_auth_method_type in databases_filter.auth_methods]
    if databases_filter.db_warnings_filter:
        if databases_filter.db_warnings_filter in (
            ArkDPADBWarning.AnyError,
            ArkDPADBWarning.NoCertificates,
        ):
            databases.items = [d for d in databases.items if not d.certificate]
        if databases_filter.db_warnings_filter in (
            ArkDPADBWarning.AnyError,
            ArkDPADBWarning.NoSecrets,
        ):
            databases.items = [d for d in databases.items if not d.secret_id]
    databases.total_count = len(databases.items)
    return databases

update_database(update_database)

Updates a database.

Parameters:

Name Type Description Default
update_database ArkDPADBUpdateDatabase

description

 required

Raises:

Type Description
ArkServiceException

description

Returns:

Name Type Description
ArkDPADBDatabase ArkDPADBDatabase

description
Source code in ark_sdk_python/services/dpa/workspaces/db/ark_dpa_db_workspace_service.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
135
136
137
138
139
140
def update_database(self, update_database: ArkDPADBUpdateDatabase) -> ArkDPADBDatabase:
    """
    Updates a database.

    Args:
        update_database (ArkDPADBUpdateDatabase): _description_

    Raises:
        ArkServiceException: _description_

    Returns:
        ArkDPADBDatabase: _description_
    """
    if update_database.name and not update_database.id:
        databases = self.list_databases_by(ArkDPADBDatabasesFilter(name=update_database.name))
        if not databases.items or len(databases.items) != 1:
            raise ArkServiceException(f'Failed to update database - name [{update_database.name}] not found')
        update_database.id = self.list_databases_by(ArkDPADBDatabasesFilter(name=update_database.name)).items[0].id
    self._logger.info(f'Updating database [{update_database.id}]')
    update_database_dict = update_database.dict(exclude={'name', 'new_name'}, exclude_none=True)
    if update_database.new_name:
        update_database_dict["name"] = update_database.new_name
    resp: Response = self.__client.put(RESOURCE_API.format(resource_id=update_database.id), json=update_database_dict)
    if resp.status_code == HTTPStatus.OK:
        try:
            return self.database(ArkDPADBGetDatabase(id=update_database.id))
        except (ValidationError, JSONDecodeError, KeyError) as ex:
            self._logger.exception(f'Failed to parse update database response [{str(ex)}] - [{resp.text}]')
            raise ArkServiceException(f'Failed to parse update database response [{str(ex)}]') from ex
    raise ArkServiceException(f'Failed to update database [{resp.text}] - [{resp.status_code}]')