aria.ops.adapter_instance

 1#  Copyright 2022 VMware, Inc.
 2#  SPDX-License-Identifier: Apache-2.0
 3from __future__ import annotations
 4
 5import sys
 6from typing import Dict
 7from typing import Optional
 8
 9from aria.ops.object import Identifier
10from aria.ops.object import Key
11from aria.ops.object import Object
12from aria.ops.pipe_utils import read_from_pipe
13from aria.ops.suite_api_client import SuiteApiClient
14from aria.ops.suite_api_client import SuiteApiConnectionParameters
15
16
17class AdapterInstance(Object):  # type: ignore
18    def __init__(self, json: dict) -> None:
19        adapter_key = json.get("adapter_key", {})
20        super().__init__(
21            Key(
22                adapter_kind=adapter_key.get("adapter_kind"),
23                object_kind=adapter_key.get("object_kind"),
24                name=adapter_key.get("name"),
25                identifiers=[
26                    Identifier(
27                        identifier.get("key"),
28                        identifier.get("value"),
29                        identifier.get("is_part_of_uniqueness"),
30                    )
31                    for identifier in adapter_key.get("identifiers", [])
32                ],
33            )
34        )
35
36        credential_config = json.get("credential_config")
37        if type(credential_config) is dict:
38            self.credential_type = credential_config.get("credential_key", None)
39            self.credentials = {
40                credential.get("key"): credential.get("value")
41                for credential in credential_config.get("credential_fields", [])
42            }
43        else:
44            self.credential_type = None
45            self.credentials = {}
46
47        cluster_connection_info = json.get("cluster_connection_info")
48        if type(cluster_connection_info) is dict:
49            self.suite_api_client = SuiteApiClient(
50                SuiteApiConnectionParameters(
51                    username=cluster_connection_info.get("user_name"),
52                    password=cluster_connection_info.get("password"),
53                    host=cluster_connection_info.get("host_name"),
54                )
55            )
56        else:
57            self.suite_api_client = None
58
59        certificate_config = json.get("certificate_config")
60        if type(certificate_config) is dict:
61            self.certificates = certificate_config.get("certificates", [])
62        else:
63            self.certificates = []
64
65        self.collection_number: Optional[int] = json.get("collection_number", None)
66        self.collection_window: Optional[Dict] = json.get("collection_window", None)
67
68    def get_credential_type(self) -> Optional[str]:
69        """Get the type (key) of credential. This is useful if an adapter supports multiple types of credentials.
70
71        :return: the type of the credential used by this adapter instance, or None if the adapter instance does not have a credential.
72        """
73        return self.credential_type  # type: ignore[no-any-return]
74
75    def get_credential_value(self, credential_key: str) -> Optional[str]:
76        """Retrieve the value of a given credential
77
78        :param credential_key: Key of the credential field
79        :return: value associated with the credential field, or None if a credential field with the given key does not exist.
80        """
81        return self.credentials.get(credential_key)
82
83    @classmethod
84    def from_input(cls, infile: str = sys.argv[-2]) -> AdapterInstance:
85        # The server always invokes methods with the input file as the second to last argument
86        return cls(read_from_pipe(infile))
class AdapterInstance(aria.ops.object.Object):
18class AdapterInstance(Object):  # type: ignore
19    def __init__(self, json: dict) -> None:
20        adapter_key = json.get("adapter_key", {})
21        super().__init__(
22            Key(
23                adapter_kind=adapter_key.get("adapter_kind"),
24                object_kind=adapter_key.get("object_kind"),
25                name=adapter_key.get("name"),
26                identifiers=[
27                    Identifier(
28                        identifier.get("key"),
29                        identifier.get("value"),
30                        identifier.get("is_part_of_uniqueness"),
31                    )
32                    for identifier in adapter_key.get("identifiers", [])
33                ],
34            )
35        )
36
37        credential_config = json.get("credential_config")
38        if type(credential_config) is dict:
39            self.credential_type = credential_config.get("credential_key", None)
40            self.credentials = {
41                credential.get("key"): credential.get("value")
42                for credential in credential_config.get("credential_fields", [])
43            }
44        else:
45            self.credential_type = None
46            self.credentials = {}
47
48        cluster_connection_info = json.get("cluster_connection_info")
49        if type(cluster_connection_info) is dict:
50            self.suite_api_client = SuiteApiClient(
51                SuiteApiConnectionParameters(
52                    username=cluster_connection_info.get("user_name"),
53                    password=cluster_connection_info.get("password"),
54                    host=cluster_connection_info.get("host_name"),
55                )
56            )
57        else:
58            self.suite_api_client = None
59
60        certificate_config = json.get("certificate_config")
61        if type(certificate_config) is dict:
62            self.certificates = certificate_config.get("certificates", [])
63        else:
64            self.certificates = []
65
66        self.collection_number: Optional[int] = json.get("collection_number", None)
67        self.collection_window: Optional[Dict] = json.get("collection_window", None)
68
69    def get_credential_type(self) -> Optional[str]:
70        """Get the type (key) of credential. This is useful if an adapter supports multiple types of credentials.
71
72        :return: the type of the credential used by this adapter instance, or None if the adapter instance does not have a credential.
73        """
74        return self.credential_type  # type: ignore[no-any-return]
75
76    def get_credential_value(self, credential_key: str) -> Optional[str]:
77        """Retrieve the value of a given credential
78
79        :param credential_key: Key of the credential field
80        :return: value associated with the credential field, or None if a credential field with the given key does not exist.
81        """
82        return self.credentials.get(credential_key)
83
84    @classmethod
85    def from_input(cls, infile: str = sys.argv[-2]) -> AdapterInstance:
86        # The server always invokes methods with the input file as the second to last argument
87        return cls(read_from_pipe(infile))

Represents an Object (resource) in vROps.

Contains Metric, Property, Event, and relationships to other Objects. Each Object is identified by a unique Key.

AdapterInstance(json: dict)
19    def __init__(self, json: dict) -> None:
20        adapter_key = json.get("adapter_key", {})
21        super().__init__(
22            Key(
23                adapter_kind=adapter_key.get("adapter_kind"),
24                object_kind=adapter_key.get("object_kind"),
25                name=adapter_key.get("name"),
26                identifiers=[
27                    Identifier(
28                        identifier.get("key"),
29                        identifier.get("value"),
30                        identifier.get("is_part_of_uniqueness"),
31                    )
32                    for identifier in adapter_key.get("identifiers", [])
33                ],
34            )
35        )
36
37        credential_config = json.get("credential_config")
38        if type(credential_config) is dict:
39            self.credential_type = credential_config.get("credential_key", None)
40            self.credentials = {
41                credential.get("key"): credential.get("value")
42                for credential in credential_config.get("credential_fields", [])
43            }
44        else:
45            self.credential_type = None
46            self.credentials = {}
47
48        cluster_connection_info = json.get("cluster_connection_info")
49        if type(cluster_connection_info) is dict:
50            self.suite_api_client = SuiteApiClient(
51                SuiteApiConnectionParameters(
52                    username=cluster_connection_info.get("user_name"),
53                    password=cluster_connection_info.get("password"),
54                    host=cluster_connection_info.get("host_name"),
55                )
56            )
57        else:
58            self.suite_api_client = None
59
60        certificate_config = json.get("certificate_config")
61        if type(certificate_config) is dict:
62            self.certificates = certificate_config.get("certificates", [])
63        else:
64            self.certificates = []
65
66        self.collection_number: Optional[int] = json.get("collection_number", None)
67        self.collection_window: Optional[Dict] = json.get("collection_window", None)

Create a new Object with a given Key.

The preferred way to create a new Object is to call the Result.object method on the Result class, which ensures that for a given key only one Object exists.

Parameters
  • key: The Key that uniquely identifies this Object
def get_credential_type(self) -> Optional[str]:
69    def get_credential_type(self) -> Optional[str]:
70        """Get the type (key) of credential. This is useful if an adapter supports multiple types of credentials.
71
72        :return: the type of the credential used by this adapter instance, or None if the adapter instance does not have a credential.
73        """
74        return self.credential_type  # type: ignore[no-any-return]

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.

def get_credential_value(self, credential_key: str) -> Optional[str]:
76    def get_credential_value(self, credential_key: str) -> Optional[str]:
77        """Retrieve the value of a given credential
78
79        :param credential_key: Key of the credential field
80        :return: value associated with the credential field, or None if a credential field with the given key does not exist.
81        """
82        return self.credentials.get(credential_key)

Retrieve the value of a given credential

Parameters
  • credential_key: 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.

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