Skip to content

Adapter instance

AdapterInstance

Bases: Object

Source code in aria/ops/adapter_instance.py 
17
18
19
20
21
22
23
24
25
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
class AdapterInstance(Object):  # type: ignore
    def __init__(self, json: dict) -> None:
        adapter_key = json.get("adapter_key", {})
        super().__init__(
            Key(
                adapter_kind=adapter_key.get("adapter_kind"),
                object_kind=adapter_key.get("object_kind"),
                name=adapter_key.get("name"),
                identifiers=[
                    Identifier(
                        identifier.get("key"),
                        identifier.get("value"),
                        identifier.get("is_part_of_uniqueness"),
                    )
                    for identifier in adapter_key.get("identifiers", [])
                ],
            )
        )

        credential_config = json.get("credential_config")
        if type(credential_config) is dict:
            self.credential_type = credential_config.get("credential_key", None)
            self.credentials = {
                credential.get("key"): credential.get("value")
                for credential in credential_config.get("credential_fields", [])
            }
        else:
            self.credential_type = None
            self.credentials = {}

        cluster_connection_info = json.get("cluster_connection_info")
        if type(cluster_connection_info) is dict:
            self.suite_api_client = SuiteApiClient(
                SuiteApiConnectionParameters(
                    username=cluster_connection_info.get("user_name"),
                    password=cluster_connection_info.get("password"),
                    host=cluster_connection_info.get("host_name"),
                )
            )
        else:
            self.suite_api_client = None

        certificate_config = json.get("certificate_config")
        if type(certificate_config) is dict:
            self.certificates = certificate_config.get("certificates", [])
        else:
            self.certificates = []

        self.collection_number: Optional[int] = json.get("collection_number", None)
        self.collection_window: Optional[Dict] = json.get("collection_window", None)

    def get_credential_type(self) -> Optional[str]:
        """Get the type (key) of credential. This is useful if an adapter supports multiple types of credentials.

        Returns:
            the type of the credential used by this adapter instance, or None if the adapter instance does not have a credential.
        """
        return self.credential_type  # type: ignore[no-any-return]

    def get_credential_value(self, credential_key: str) -> Optional[str]:
        """Retrieve the value of a given credential

        Args:
            credential_key (str): Key of the credential field

        Returns:
            value associated with the credential field, or None if a credential field with the given key does not exist.
        """
        return self.credentials.get(credential_key)

    @classmethod
    def from_input(cls, infile: str = sys.argv[-2]) -> AdapterInstance:
        # The server always invokes methods with the input file as the second to last argument
        return cls(read_from_pipe(infile))

get_credential_type()

Get the type (key) of credential. This is useful if an adapter supports multiple types of credentials.

Returns:

Type Description
Optional[str]

the type of the credential used by this adapter instance, or None if the adapter instance does not have a credential.
Source code in aria/ops/adapter_instance.py 
68
69
70
71
72
73
74
def get_credential_type(self) -> Optional[str]:
    """Get the type (key) of credential. This is useful if an adapter supports multiple types of credentials.

    Returns:
        the type of the credential used by this adapter instance, or None if the adapter instance does not have a credential.
    """
    return self.credential_type  # type: ignore[no-any-return]

get_credential_value(credential_key)

Retrieve the value of a given credential

Parameters:

Name Type Description Default
credential_key str

Key of the credential field

 required

Returns:

Type Description
Optional[str]

value associated with the credential field, or None if a credential field with the given key does not exist.
Source code in aria/ops/adapter_instance.py 
76
77
78
79
80
81
82
83
84
85
def get_credential_value(self, credential_key: str) -> Optional[str]:
    """Retrieve the value of a given credential

    Args:
        credential_key (str): Key of the credential field

    Returns:
        value associated with the credential field, or None if a credential field with the given key does not exist.
    """
    return self.credentials.get(credential_key)