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))
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.
Inherited Members
- aria.ops.object.Object
- get_key
- adapter_type
- object_type
- get_identifier_value
- add_metric
- add_metrics
- with_metric
- get_metric
- get_metric_values
- get_last_metric_value
- add_property
- add_properties
- with_property
- get_property
- get_property_values
- get_last_property_value
- add_event
- add_events
- with_event
- add_parent
- add_parents
- get_parents
- add_child
- add_children
- get_children
- has_content
- get_json