Skip to content

ark_identity_connectors_service

ArkIdentityConnectorsService

Bases: ArkIdentityBaseService

Source code in ark_sdk_python/services/identity/connectors/ark_identity_connectors_service.py 
 26
 27
 28
 29
 30
 31
 32
 33
 34
 35
 36
 37
 38
 39
 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
class ArkIdentityConnectorsService(ArkIdentityBaseService):
    def list_connectors(self) -> List[ArkIdentityConnectorInfo]:
        """
        Lists all identity connectors on the tenant

        Returns:
            List[ArkIdentityConnectorInfo]: _description_
        """
        self._logger.info('Listing all identity connectors')
        response: Response = self._client.post(
            f'{self._url_prefix}{REDROCK_QUERY}',
            json={"Script": "Select * from Proxy"},
        )
        if response.status_code != HTTPStatus.OK:
            raise ArkServiceException(f'Failed to retrieve identity connectors [{response.text}] - [{response.status_code}]')
        try:
            query_result = response.json()
            if not query_result['success']:
                raise ArkServiceException('Failed to retrieve identity connectors')
            if len(query_result['Result']["Results"]) == 0:
                return []
            return TypeAdapter(List[ArkIdentityConnectorInfo]).validate_python([r['Row'] for r in query_result['Result']["Results"]])
        except (ValidationError, JSONDecodeError, KeyError) as ex:
            self._logger.exception(f'Failed to retrieve identity connectors [{str(ex)}] - [{response.text}]')
            raise ArkServiceException(f'Failed to retrieve identity connectors [{str(ex)}]') from ex

    def list_connectors_by(self, connectors_filter: ArkIdentityConnectorsFilter) -> List[ArkIdentityConnectorInfo]:
        """
        Lists identity connectors on the tenant by filters

        Args:
            connectors_filter (ArkIdentityConnectorsFilter): _description_

        Returns:
            List[ArkIdentityConnectorInfo]: _description_
        """
        self._logger.info(f'Listing identity connectors by filters [{connectors_filter}]')
        connectors = self.list_connectors()

        # Filter by connector online / offline
        if connectors_filter.online is not None:
            connectors = [c for c in connectors if c.online == connectors_filter.online]

        # Filter by forest
        if connectors_filter.forest:
            connectors = [c for c in connectors if fnmatch(c.forest, connectors_filter.forest)]

        # Filter by dns
        if connectors_filter.dns:
            connectors = [c for c in connectors if fnmatch(c.dns_host_name, connectors_filter.dns)]

        # Filter by machine name
        if connectors_filter.machine_name:
            connectors = [c for c in connectors if fnmatch(c.machine_name, connectors_filter.machine_name)]

        # Filter by customer name
        if connectors_filter.customer_name:
            connectors = [c for c in connectors if fnmatch(c.customer_name, connectors_filter.customer_name)]

        # Filter by forest
        if connectors_filter.version:
            connectors = [c for c in connectors if fnmatch(c.version, connectors_filter.version)]

        return connectors

    def connector(self, get_connector: ArkIdentityGetConnector) -> ArkIdentityConnectorInfo:
        """
        Retrieves a connector by id

        Args:
            get_connector (ArkIdentityGetConnector): _description_

        Returns:
            ArkIdentityConnectorInfo: _description_
        """
        self._logger.info(f'Retrieving identity connector by id [{get_connector.connector_id}]')
        response: Response = self._client.post(
            f'{self._url_prefix}{REDROCK_QUERY}',
            json={"Script": f"Select * from Proxy WHERE ID='{get_connector.connector_id}'"},
        )
        if response.status_code != HTTPStatus.OK:
            raise ArkServiceException(f'Failed to retrieve identity connector 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 identity connector by id')
            return ArkIdentityConnectorInfo.model_validate(query_result['Result']["Results"][0]['Row'])
        except (ValidationError, JSONDecodeError, KeyError) as ex:
            self._logger.exception(f'Failed to retrieve identity connector by id [{str(ex)}] - [{response.text}]')
            raise ArkServiceException(f'Failed to retrieve identity connector by id [{str(ex)}]') from ex

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

connector(get_connector)

Retrieves a connector by id

Parameters:

Name Type Description Default
get_connector ArkIdentityGetConnector

description

 required

Returns:

Name Type Description
ArkIdentityConnectorInfo ArkIdentityConnectorInfo

description
Source code in ark_sdk_python/services/identity/connectors/ark_identity_connectors_service.py 
 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
def connector(self, get_connector: ArkIdentityGetConnector) -> ArkIdentityConnectorInfo:
    """
    Retrieves a connector by id

    Args:
        get_connector (ArkIdentityGetConnector): _description_

    Returns:
        ArkIdentityConnectorInfo: _description_
    """
    self._logger.info(f'Retrieving identity connector by id [{get_connector.connector_id}]')
    response: Response = self._client.post(
        f'{self._url_prefix}{REDROCK_QUERY}',
        json={"Script": f"Select * from Proxy WHERE ID='{get_connector.connector_id}'"},
    )
    if response.status_code != HTTPStatus.OK:
        raise ArkServiceException(f'Failed to retrieve identity connector 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 identity connector by id')
        return ArkIdentityConnectorInfo.model_validate(query_result['Result']["Results"][0]['Row'])
    except (ValidationError, JSONDecodeError, KeyError) as ex:
        self._logger.exception(f'Failed to retrieve identity connector by id [{str(ex)}] - [{response.text}]')
        raise ArkServiceException(f'Failed to retrieve identity connector by id [{str(ex)}]') from ex

list_connectors()

Lists all identity connectors on the tenant

Returns:

Type Description
List[ArkIdentityConnectorInfo]

List[ArkIdentityConnectorInfo]: description
Source code in ark_sdk_python/services/identity/connectors/ark_identity_connectors_service.py 
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
def list_connectors(self) -> List[ArkIdentityConnectorInfo]:
    """
    Lists all identity connectors on the tenant

    Returns:
        List[ArkIdentityConnectorInfo]: _description_
    """
    self._logger.info('Listing all identity connectors')
    response: Response = self._client.post(
        f'{self._url_prefix}{REDROCK_QUERY}',
        json={"Script": "Select * from Proxy"},
    )
    if response.status_code != HTTPStatus.OK:
        raise ArkServiceException(f'Failed to retrieve identity connectors [{response.text}] - [{response.status_code}]')
    try:
        query_result = response.json()
        if not query_result['success']:
            raise ArkServiceException('Failed to retrieve identity connectors')
        if len(query_result['Result']["Results"]) == 0:
            return []
        return TypeAdapter(List[ArkIdentityConnectorInfo]).validate_python([r['Row'] for r in query_result['Result']["Results"]])
    except (ValidationError, JSONDecodeError, KeyError) as ex:
        self._logger.exception(f'Failed to retrieve identity connectors [{str(ex)}] - [{response.text}]')
        raise ArkServiceException(f'Failed to retrieve identity connectors [{str(ex)}]') from ex

list_connectors_by(connectors_filter)

Lists identity connectors on the tenant by filters

Parameters:

Name Type Description Default
connectors_filter ArkIdentityConnectorsFilter

description

 required

Returns:

Type Description
List[ArkIdentityConnectorInfo]

List[ArkIdentityConnectorInfo]: description
Source code in ark_sdk_python/services/identity/connectors/ark_identity_connectors_service.py 
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
def list_connectors_by(self, connectors_filter: ArkIdentityConnectorsFilter) -> List[ArkIdentityConnectorInfo]:
    """
    Lists identity connectors on the tenant by filters

    Args:
        connectors_filter (ArkIdentityConnectorsFilter): _description_

    Returns:
        List[ArkIdentityConnectorInfo]: _description_
    """
    self._logger.info(f'Listing identity connectors by filters [{connectors_filter}]')
    connectors = self.list_connectors()

    # Filter by connector online / offline
    if connectors_filter.online is not None:
        connectors = [c for c in connectors if c.online == connectors_filter.online]

    # Filter by forest
    if connectors_filter.forest:
        connectors = [c for c in connectors if fnmatch(c.forest, connectors_filter.forest)]

    # Filter by dns
    if connectors_filter.dns:
        connectors = [c for c in connectors if fnmatch(c.dns_host_name, connectors_filter.dns)]

    # Filter by machine name
    if connectors_filter.machine_name:
        connectors = [c for c in connectors if fnmatch(c.machine_name, connectors_filter.machine_name)]

    # Filter by customer name
    if connectors_filter.customer_name:
        connectors = [c for c in connectors if fnmatch(c.customer_name, connectors_filter.customer_name)]

    # Filter by forest
    if connectors_filter.version:
        connectors = [c for c in connectors if fnmatch(c.version, connectors_filter.version)]

    return connectors