aria.ops.result
1# Copyright 2022 VMware, Inc. 2# SPDX-License-Identifier: Apache-2.0 3import sys 4from enum import auto 5from typing import List 6from typing import NewType 7from typing import Optional 8 9from aenum import Enum 10from aria.ops.definition.adapter_definition import AdapterDefinition 11from aria.ops.object import Identifier 12from aria.ops.object import Key 13from aria.ops.object import Object 14from aria.ops.pipe_utils import write_to_pipe 15 16 17class ObjectKeyAlreadyExistsException(Exception): 18 """Exception when two objects with the same Key are added to the same 19 :class:`Result`""" 20 21 pass 22 23 24class TestResult: 25 """Class for managing the results of an adapter instance connection test""" 26 27 def __init__(self) -> None: 28 """Initializes a Result""" 29 self._error_message: Optional[str] = None 30 31 def is_success(self) -> bool: 32 """ 33 34 :return: True if the TestResult represents a successful test 35 """ 36 return self._error_message is None 37 38 def with_error(self, error_message: str) -> None: 39 """Set the adapter instance connection test to failed, and display the given 40 error message. 41 42 If this method is called multiple times, only the most recent error message 43 will be recorded. If error_message is set, the test is considered failed. 44 45 :param error_message: A string containing the error message 46 :return: None 47 """ 48 self._error_message = error_message 49 50 def get_json(self) -> dict: 51 """Get a JSON representation of this TestResult 52 53 Returns a JSON representation of this TestResult in the format required by 54 Aria Operations, indicating either a successful test, or a failed test with 55 error message. 56 57 :return: A JSON representation of this TestResult 58 """ 59 if self.is_success(): 60 return {} 61 else: 62 return {"errorMessage": self._error_message} 63 64 def send_results(self, output_pipe: str = sys.argv[-1]) -> None: 65 """Opens the output pipe and sends results directly back to the server 66 67 This method can only be called once per collection. 68 """ 69 # The server always invokes methods with the output file as the last argument 70 write_to_pipe(output_pipe, self.get_json()) 71 72 73class EndpointResult: 74 """Class for managing the results of an adapter instance get endpoint URLs call 75 76 The result of this should be a set of urls that the adapter will connect to. 77 Aria Operations will then attempt to connect to each of these urls securely, 78 and prompt the user to accept or reject the certificate presented by each URL. 79 80 """ 81 82 def __init__(self) -> None: 83 """Initializes an EndpointResult""" 84 self.endpoints: list[str] = [] 85 86 def with_endpoint(self, endpoint_url: str) -> None: 87 """Adds an endpoint to the list of endpoints Aria Operations will test for 88 certificate validation. 89 90 If this method is called multiple times, each url will be called by Aria 91 Operations. 92 93 :param endpoint_url: A string containing the url 94 95 :return: None 96 """ 97 self.endpoints.append(endpoint_url) 98 99 def get_json(self) -> dict: 100 """Get a JSON representation of this EndpointResult 101 102 Returns a JSON representation of this EndpointResult in the format required 103 by Aria Operations. 104 105 :return: A JSON representation of this EndpointResult 106 """ 107 return {"endpointUrls": self.endpoints} 108 109 def send_results(self, output_pipe: str = sys.argv[-1]) -> None: 110 """Opens the output pipe and sends results directly back to the server 111 112 This method can only be called once per collection. 113 """ 114 # The server always invokes methods with the output file as the last argument 115 write_to_pipe(output_pipe, self.get_json()) 116 117 118RelationshipUpdateMode = NewType("RelationshipUpdateMode", int) 119 120 121class RelationshipUpdateModes(Enum): # type: ignore 122 """ 123 If 'update_relationships' is 'ALL', all relationships between objects are 124 returned. This mode will remove any currently-existing relationships in VMware 125 Aria Operations that are not present in the Result. 126 127 If 'update_relationships' is 'NONE', no relationships will be returned, even if 128 there are relationships between objects in the Result. All currently-existing 129 relationships in VMware Aria Operations will be preserved. 130 131 If 'update_relationships' is 'AUTO' (or not explicitly set), then the mode will 132 behave like 'ALL' if any object in the Result has at least one relationship, 133 otherwise the mode will behave like 'NONE' if no objects have any relationships in 134 the Result. This default behavior makes it easy to skip collecting all relationships 135 for a collection without overwriting previously-collected relationships, e.g., for 136 performance reasons. 137 138 If 'update_relationships' is 'PER_OBJECT', then only objects with updated 139 relationships will be returned. This is similar to 'AUTO' except that if an 140 object's child relationships have not been updated/set (by calling 'add_child' or 141 'add_children'), existing child relationships in VMware Aria Operations will be 142 preserved. This means that to remove all relationships from an object 143 (without setting any new relationships), the adapter must call 'add_children' on 144 the object with an empty collection of children. This mode is useful for updating a 145 subset of objects' relationships in a collection, but requires more care to 146 ensure relationships are removed when appropriate. 147 """ 148 149 ALL = RelationshipUpdateMode(1) 150 NONE = RelationshipUpdateMode(2) 151 AUTO = RelationshipUpdateMode(3) 152 PER_OBJECT = RelationshipUpdateMode(4) 153 154 155class CollectResult: 156 """Class for managing a collection of Aria Operations Objects""" 157 158 def __init__( 159 self, 160 obj_list: Optional[list[Object]] = None, 161 target_definition: AdapterDefinition = None, 162 ) -> None: 163 """Initializes a Result 164 165 A result contains objects, which can be added at initialization or later. 166 Each object has a key containing one or more identifiers plus the object type 167 and adapter type. Keys must be unique across objects in a Result. 168 169 :param obj_list: an optional list of objects to send to Aria Operations. 170 Objects can be added later using add_object 171 172 :param target_definition: an optional description of the returned objects, 173 used for validation purposes 174 """ 175 self.objects: dict[Key, Object] = {} 176 if type(obj_list) is list: 177 self.add_objects(obj_list) 178 self.definition: AdapterDefinition = target_definition 179 self.adapter_type = None 180 if self.definition: 181 self.adapter_type = self.definition.key 182 self._error_message: Optional[str] = None 183 self.update_relationships: RelationshipUpdateMode = RelationshipUpdateModes.AUTO 184 185 def _object_is_external(self, obj: Object) -> bool: 186 return bool(self.adapter_type) and not obj.adapter_type() == self.adapter_type 187 188 def object( 189 self, 190 adapter_kind: str, 191 object_kind: str, 192 name: str, 193 identifiers: Optional[list[Identifier]] = None, 194 ) -> Object: 195 """Get or create the object with key specified by adapter_kind, object_kind, 196 name, and identifiers. 197 198 This is the preferred method for creating new Objects. If this method is used 199 exclusively, all object references with the same key will point to the same 200 object. 201 202 If an object with the same key already exists in the result, return that 203 object, otherwise create a new object, add it to the result, and return it. 204 See discussion on keys in the documentation for the :class:`object.Key` class. 205 206 If this method is used to create an object, it does not need to be added 207 later using `add_object` (or `add_objects`) 208 209 :param adapter_kind: The adapter kind of the object 210 :param object_kind: The resource kind of the object 211 :param name: The name of the object 212 :param identifiers: 213 :return: The object with the given key 214 """ 215 obj = Object(Key(adapter_kind, object_kind, name, identifiers)) 216 return self.objects.setdefault(obj.get_key(), obj) 217 218 def get_object(self, obj_key: Key) -> Optional[Object]: 219 """Get and return the object corresponding to the given key, if it exists 220 221 :param obj_key: The object key to search for 222 :return: The object with the given key, or None if the key is not in the result 223 """ 224 return self.objects.get(obj_key, None) 225 226 def get_objects_by_type( 227 self, object_type: str, adapter_type: Optional[str] = None 228 ) -> List[Object]: 229 """Returns all objects with the given type. If adapter_type is present, 230 the objects must also be from the given adapter type. 231 232 :param object_type: The object type to return 233 :param adapter_type: The adapter type of the objects to return 234 :return: A list of objects matching the object type and adapter type 235 """ 236 return [ 237 obj 238 for obj in self.objects.values() 239 if obj.adapter_type() == object_type 240 and (adapter_type is None or adapter_type == obj.adapter_type()) 241 ] 242 243 def add_object(self, obj: Object) -> Object: 244 """Adds the given object to the Result and returns it. 245 246 Adds the given object to the Result and returns it. A different object with 247 the same key cannot already exist in the Result. If it does, 248 an :class:`ObjectKeyAlreadyExistsException` will be raised. 249 250 :param obj: An object to add to the Result 251 :return: The object 252 :raises: ObjectKeyAlreadyExistsException if a different object with the same key 253 already exists in the Result 254 """ 255 o = self.objects.setdefault(obj.get_key(), obj) 256 if o is obj: 257 return o 258 raise ObjectKeyAlreadyExistsException( 259 f"A different object with key {obj.get_key()} already exists." 260 ) 261 262 def add_objects(self, obj_list: list[Object]) -> None: 263 """Adds the given objects to the Result and returns it. 264 265 Adds the given objects to the Result. A different object with the same key 266 cannot already exist in the Result. If it does, an 267 :class:`ObjectKeyAlreadyExistsException` will be raised. 268 269 :param obj_list: A list of objects to add to the Result 270 :return: The object 271 :raises: ObjectKeyAlreadyExistsException if a different object with the same 272 key already exists in the Result 273 """ 274 for obj in obj_list: 275 self.add_object(obj) 276 277 def with_error(self, error_message: str) -> None: 278 """Set the Adapter Instance to an error state with the provided message. 279 280 If this method is called multiple times, only the most recent error message 281 will be recorded. If error_message is set, no results (objects, relationships) 282 will be returned. 283 284 :param error_message: A string containing the error message 285 :return: None 286 """ 287 self._error_message = error_message 288 289 def get_json(self) -> dict: 290 """Get a JSON representation of this Result 291 292 Returns a JSON representation of this Result in the format required by Aria 293 Operations. The representation includes all objects (including the object's 294 events, properties, and metrics) in the Result. Relationships are returned 295 following the update_relationships flag (See RelationshipUpdateMode). 296 297 :return: A JSON representation of this Result 298 """ 299 if self._error_message is None: 300 result = { 301 "result": [ 302 obj.get_json() 303 for obj in self.objects.values() 304 if not self._object_is_external(obj) or obj.has_content() 305 ], 306 "relationships": [], 307 "nonExistingObjects": [], 308 } 309 if ( 310 self.update_relationships == RelationshipUpdateModes.ALL 311 or self.update_relationships == RelationshipUpdateModes.PER_OBJECT 312 or ( 313 self.update_relationships == RelationshipUpdateModes.AUTO 314 and any([obj._updated_children for obj in self.objects.values()]) 315 ) 316 ): 317 result.update( 318 { 319 "relationships": [ 320 { 321 "parent": obj.get_key().get_json(), 322 "children": [ 323 child_key.get_json() 324 for child_key in obj.get_children() 325 ], 326 } 327 for obj in self.objects.values() 328 if ( 329 self.update_relationships 330 == RelationshipUpdateModes.PER_OBJECT 331 and obj._updated_children 332 ) 333 or not self.update_relationships 334 == RelationshipUpdateModes.PER_OBJECT 335 ], 336 } 337 ) 338 return result 339 else: 340 return {"errorMessage": self._error_message} 341 342 def send_results(self, output_pipe: str = sys.argv[-1]) -> None: 343 """Opens the output pipe and sends results directly back to the server 344 345 This method can only be called once per collection. 346 """ 347 # The server always invokes methods with the output file as the last argument 348 write_to_pipe(output_pipe, self.get_json())
18class ObjectKeyAlreadyExistsException(Exception): 19 """Exception when two objects with the same Key are added to the same 20 :class:`Result`""" 21 22 pass
Exception when two objects with the same Key are added to the same
Result
Inherited Members
- builtins.Exception
- Exception
- builtins.BaseException
- with_traceback
- add_note
25class TestResult: 26 """Class for managing the results of an adapter instance connection test""" 27 28 def __init__(self) -> None: 29 """Initializes a Result""" 30 self._error_message: Optional[str] = None 31 32 def is_success(self) -> bool: 33 """ 34 35 :return: True if the TestResult represents a successful test 36 """ 37 return self._error_message is None 38 39 def with_error(self, error_message: str) -> None: 40 """Set the adapter instance connection test to failed, and display the given 41 error message. 42 43 If this method is called multiple times, only the most recent error message 44 will be recorded. If error_message is set, the test is considered failed. 45 46 :param error_message: A string containing the error message 47 :return: None 48 """ 49 self._error_message = error_message 50 51 def get_json(self) -> dict: 52 """Get a JSON representation of this TestResult 53 54 Returns a JSON representation of this TestResult in the format required by 55 Aria Operations, indicating either a successful test, or a failed test with 56 error message. 57 58 :return: A JSON representation of this TestResult 59 """ 60 if self.is_success(): 61 return {} 62 else: 63 return {"errorMessage": self._error_message} 64 65 def send_results(self, output_pipe: str = sys.argv[-1]) -> None: 66 """Opens the output pipe and sends results directly back to the server 67 68 This method can only be called once per collection. 69 """ 70 # The server always invokes methods with the output file as the last argument 71 write_to_pipe(output_pipe, self.get_json())
Class for managing the results of an adapter instance connection test
28 def __init__(self) -> None: 29 """Initializes a Result""" 30 self._error_message: Optional[str] = None
Initializes a Result
32 def is_success(self) -> bool: 33 """ 34 35 :return: True if the TestResult represents a successful test 36 """ 37 return self._error_message is None
Returns
True if the TestResult represents a successful test
39 def with_error(self, error_message: str) -> None: 40 """Set the adapter instance connection test to failed, and display the given 41 error message. 42 43 If this method is called multiple times, only the most recent error message 44 will be recorded. If error_message is set, the test is considered failed. 45 46 :param error_message: A string containing the error message 47 :return: None 48 """ 49 self._error_message = error_message
Set the adapter instance connection test to failed, and display the given error message.
If this method is called multiple times, only the most recent error message will be recorded. If error_message is set, the test is considered failed.
Parameters
- error_message: A string containing the error message
Returns
None
51 def get_json(self) -> dict: 52 """Get a JSON representation of this TestResult 53 54 Returns a JSON representation of this TestResult in the format required by 55 Aria Operations, indicating either a successful test, or a failed test with 56 error message. 57 58 :return: A JSON representation of this TestResult 59 """ 60 if self.is_success(): 61 return {} 62 else: 63 return {"errorMessage": self._error_message}
Get a JSON representation of this TestResult
Returns a JSON representation of this TestResult in the format required by Aria Operations, indicating either a successful test, or a failed test with error message.
Returns
A JSON representation of this TestResult
65 def send_results(self, output_pipe: str = sys.argv[-1]) -> None: 66 """Opens the output pipe and sends results directly back to the server 67 68 This method can only be called once per collection. 69 """ 70 # The server always invokes methods with the output file as the last argument 71 write_to_pipe(output_pipe, self.get_json())
Opens the output pipe and sends results directly back to the server
This method can only be called once per collection.
74class EndpointResult: 75 """Class for managing the results of an adapter instance get endpoint URLs call 76 77 The result of this should be a set of urls that the adapter will connect to. 78 Aria Operations will then attempt to connect to each of these urls securely, 79 and prompt the user to accept or reject the certificate presented by each URL. 80 81 """ 82 83 def __init__(self) -> None: 84 """Initializes an EndpointResult""" 85 self.endpoints: list[str] = [] 86 87 def with_endpoint(self, endpoint_url: str) -> None: 88 """Adds an endpoint to the list of endpoints Aria Operations will test for 89 certificate validation. 90 91 If this method is called multiple times, each url will be called by Aria 92 Operations. 93 94 :param endpoint_url: A string containing the url 95 96 :return: None 97 """ 98 self.endpoints.append(endpoint_url) 99 100 def get_json(self) -> dict: 101 """Get a JSON representation of this EndpointResult 102 103 Returns a JSON representation of this EndpointResult in the format required 104 by Aria Operations. 105 106 :return: A JSON representation of this EndpointResult 107 """ 108 return {"endpointUrls": self.endpoints} 109 110 def send_results(self, output_pipe: str = sys.argv[-1]) -> None: 111 """Opens the output pipe and sends results directly back to the server 112 113 This method can only be called once per collection. 114 """ 115 # The server always invokes methods with the output file as the last argument 116 write_to_pipe(output_pipe, self.get_json())
Class for managing the results of an adapter instance get endpoint URLs call
The result of this should be a set of urls that the adapter will connect to. Aria Operations will then attempt to connect to each of these urls securely, and prompt the user to accept or reject the certificate presented by each URL.
83 def __init__(self) -> None: 84 """Initializes an EndpointResult""" 85 self.endpoints: list[str] = []
Initializes an EndpointResult
87 def with_endpoint(self, endpoint_url: str) -> None: 88 """Adds an endpoint to the list of endpoints Aria Operations will test for 89 certificate validation. 90 91 If this method is called multiple times, each url will be called by Aria 92 Operations. 93 94 :param endpoint_url: A string containing the url 95 96 :return: None 97 """ 98 self.endpoints.append(endpoint_url)
Adds an endpoint to the list of endpoints Aria Operations will test for certificate validation.
If this method is called multiple times, each url will be called by Aria Operations.
Parameters
- endpoint_url: A string containing the url
Returns
None
100 def get_json(self) -> dict: 101 """Get a JSON representation of this EndpointResult 102 103 Returns a JSON representation of this EndpointResult in the format required 104 by Aria Operations. 105 106 :return: A JSON representation of this EndpointResult 107 """ 108 return {"endpointUrls": self.endpoints}
Get a JSON representation of this EndpointResult
Returns a JSON representation of this EndpointResult in the format required by Aria Operations.
Returns
A JSON representation of this EndpointResult
110 def send_results(self, output_pipe: str = sys.argv[-1]) -> None: 111 """Opens the output pipe and sends results directly back to the server 112 113 This method can only be called once per collection. 114 """ 115 # The server always invokes methods with the output file as the last argument 116 write_to_pipe(output_pipe, self.get_json())
Opens the output pipe and sends results directly back to the server
This method can only be called once per collection.
122class RelationshipUpdateModes(Enum): # type: ignore 123 """ 124 If 'update_relationships' is 'ALL', all relationships between objects are 125 returned. This mode will remove any currently-existing relationships in VMware 126 Aria Operations that are not present in the Result. 127 128 If 'update_relationships' is 'NONE', no relationships will be returned, even if 129 there are relationships between objects in the Result. All currently-existing 130 relationships in VMware Aria Operations will be preserved. 131 132 If 'update_relationships' is 'AUTO' (or not explicitly set), then the mode will 133 behave like 'ALL' if any object in the Result has at least one relationship, 134 otherwise the mode will behave like 'NONE' if no objects have any relationships in 135 the Result. This default behavior makes it easy to skip collecting all relationships 136 for a collection without overwriting previously-collected relationships, e.g., for 137 performance reasons. 138 139 If 'update_relationships' is 'PER_OBJECT', then only objects with updated 140 relationships will be returned. This is similar to 'AUTO' except that if an 141 object's child relationships have not been updated/set (by calling 'add_child' or 142 'add_children'), existing child relationships in VMware Aria Operations will be 143 preserved. This means that to remove all relationships from an object 144 (without setting any new relationships), the adapter must call 'add_children' on 145 the object with an empty collection of children. This mode is useful for updating a 146 subset of objects' relationships in a collection, but requires more care to 147 ensure relationships are removed when appropriate. 148 """ 149 150 ALL = RelationshipUpdateMode(1) 151 NONE = RelationshipUpdateMode(2) 152 AUTO = RelationshipUpdateMode(3) 153 PER_OBJECT = RelationshipUpdateMode(4)
If 'update_relationships' is 'ALL', all relationships between objects are returned. This mode will remove any currently-existing relationships in VMware Aria Operations that are not present in the Result.
If 'update_relationships' is 'NONE', no relationships will be returned, even if there are relationships between objects in the Result. All currently-existing relationships in VMware Aria Operations will be preserved.
If 'update_relationships' is 'AUTO' (or not explicitly set), then the mode will behave like 'ALL' if any object in the Result has at least one relationship, otherwise the mode will behave like 'NONE' if no objects have any relationships in the Result. This default behavior makes it easy to skip collecting all relationships for a collection without overwriting previously-collected relationships, e.g., for performance reasons.
If 'update_relationships' is 'PER_OBJECT', then only objects with updated relationships will be returned. This is similar to 'AUTO' except that if an object's child relationships have not been updated/set (by calling 'add_child' or 'add_children'), existing child relationships in VMware Aria Operations will be preserved. This means that to remove all relationships from an object (without setting any new relationships), the adapter must call 'add_children' on the object with an empty collection of children. This mode is useful for updating a subset of objects' relationships in a collection, but requires more care to ensure relationships are removed when appropriate.
156class CollectResult: 157 """Class for managing a collection of Aria Operations Objects""" 158 159 def __init__( 160 self, 161 obj_list: Optional[list[Object]] = None, 162 target_definition: AdapterDefinition = None, 163 ) -> None: 164 """Initializes a Result 165 166 A result contains objects, which can be added at initialization or later. 167 Each object has a key containing one or more identifiers plus the object type 168 and adapter type. Keys must be unique across objects in a Result. 169 170 :param obj_list: an optional list of objects to send to Aria Operations. 171 Objects can be added later using add_object 172 173 :param target_definition: an optional description of the returned objects, 174 used for validation purposes 175 """ 176 self.objects: dict[Key, Object] = {} 177 if type(obj_list) is list: 178 self.add_objects(obj_list) 179 self.definition: AdapterDefinition = target_definition 180 self.adapter_type = None 181 if self.definition: 182 self.adapter_type = self.definition.key 183 self._error_message: Optional[str] = None 184 self.update_relationships: RelationshipUpdateMode = RelationshipUpdateModes.AUTO 185 186 def _object_is_external(self, obj: Object) -> bool: 187 return bool(self.adapter_type) and not obj.adapter_type() == self.adapter_type 188 189 def object( 190 self, 191 adapter_kind: str, 192 object_kind: str, 193 name: str, 194 identifiers: Optional[list[Identifier]] = None, 195 ) -> Object: 196 """Get or create the object with key specified by adapter_kind, object_kind, 197 name, and identifiers. 198 199 This is the preferred method for creating new Objects. If this method is used 200 exclusively, all object references with the same key will point to the same 201 object. 202 203 If an object with the same key already exists in the result, return that 204 object, otherwise create a new object, add it to the result, and return it. 205 See discussion on keys in the documentation for the :class:`object.Key` class. 206 207 If this method is used to create an object, it does not need to be added 208 later using `add_object` (or `add_objects`) 209 210 :param adapter_kind: The adapter kind of the object 211 :param object_kind: The resource kind of the object 212 :param name: The name of the object 213 :param identifiers: 214 :return: The object with the given key 215 """ 216 obj = Object(Key(adapter_kind, object_kind, name, identifiers)) 217 return self.objects.setdefault(obj.get_key(), obj) 218 219 def get_object(self, obj_key: Key) -> Optional[Object]: 220 """Get and return the object corresponding to the given key, if it exists 221 222 :param obj_key: The object key to search for 223 :return: The object with the given key, or None if the key is not in the result 224 """ 225 return self.objects.get(obj_key, None) 226 227 def get_objects_by_type( 228 self, object_type: str, adapter_type: Optional[str] = None 229 ) -> List[Object]: 230 """Returns all objects with the given type. If adapter_type is present, 231 the objects must also be from the given adapter type. 232 233 :param object_type: The object type to return 234 :param adapter_type: The adapter type of the objects to return 235 :return: A list of objects matching the object type and adapter type 236 """ 237 return [ 238 obj 239 for obj in self.objects.values() 240 if obj.adapter_type() == object_type 241 and (adapter_type is None or adapter_type == obj.adapter_type()) 242 ] 243 244 def add_object(self, obj: Object) -> Object: 245 """Adds the given object to the Result and returns it. 246 247 Adds the given object to the Result and returns it. A different object with 248 the same key cannot already exist in the Result. If it does, 249 an :class:`ObjectKeyAlreadyExistsException` will be raised. 250 251 :param obj: An object to add to the Result 252 :return: The object 253 :raises: ObjectKeyAlreadyExistsException if a different object with the same key 254 already exists in the Result 255 """ 256 o = self.objects.setdefault(obj.get_key(), obj) 257 if o is obj: 258 return o 259 raise ObjectKeyAlreadyExistsException( 260 f"A different object with key {obj.get_key()} already exists." 261 ) 262 263 def add_objects(self, obj_list: list[Object]) -> None: 264 """Adds the given objects to the Result and returns it. 265 266 Adds the given objects to the Result. A different object with the same key 267 cannot already exist in the Result. If it does, an 268 :class:`ObjectKeyAlreadyExistsException` will be raised. 269 270 :param obj_list: A list of objects to add to the Result 271 :return: The object 272 :raises: ObjectKeyAlreadyExistsException if a different object with the same 273 key already exists in the Result 274 """ 275 for obj in obj_list: 276 self.add_object(obj) 277 278 def with_error(self, error_message: str) -> None: 279 """Set the Adapter Instance to an error state with the provided message. 280 281 If this method is called multiple times, only the most recent error message 282 will be recorded. If error_message is set, no results (objects, relationships) 283 will be returned. 284 285 :param error_message: A string containing the error message 286 :return: None 287 """ 288 self._error_message = error_message 289 290 def get_json(self) -> dict: 291 """Get a JSON representation of this Result 292 293 Returns a JSON representation of this Result in the format required by Aria 294 Operations. The representation includes all objects (including the object's 295 events, properties, and metrics) in the Result. Relationships are returned 296 following the update_relationships flag (See RelationshipUpdateMode). 297 298 :return: A JSON representation of this Result 299 """ 300 if self._error_message is None: 301 result = { 302 "result": [ 303 obj.get_json() 304 for obj in self.objects.values() 305 if not self._object_is_external(obj) or obj.has_content() 306 ], 307 "relationships": [], 308 "nonExistingObjects": [], 309 } 310 if ( 311 self.update_relationships == RelationshipUpdateModes.ALL 312 or self.update_relationships == RelationshipUpdateModes.PER_OBJECT 313 or ( 314 self.update_relationships == RelationshipUpdateModes.AUTO 315 and any([obj._updated_children for obj in self.objects.values()]) 316 ) 317 ): 318 result.update( 319 { 320 "relationships": [ 321 { 322 "parent": obj.get_key().get_json(), 323 "children": [ 324 child_key.get_json() 325 for child_key in obj.get_children() 326 ], 327 } 328 for obj in self.objects.values() 329 if ( 330 self.update_relationships 331 == RelationshipUpdateModes.PER_OBJECT 332 and obj._updated_children 333 ) 334 or not self.update_relationships 335 == RelationshipUpdateModes.PER_OBJECT 336 ], 337 } 338 ) 339 return result 340 else: 341 return {"errorMessage": self._error_message} 342 343 def send_results(self, output_pipe: str = sys.argv[-1]) -> None: 344 """Opens the output pipe and sends results directly back to the server 345 346 This method can only be called once per collection. 347 """ 348 # The server always invokes methods with the output file as the last argument 349 write_to_pipe(output_pipe, self.get_json())
Class for managing a collection of Aria Operations Objects
159 def __init__( 160 self, 161 obj_list: Optional[list[Object]] = None, 162 target_definition: AdapterDefinition = None, 163 ) -> None: 164 """Initializes a Result 165 166 A result contains objects, which can be added at initialization or later. 167 Each object has a key containing one or more identifiers plus the object type 168 and adapter type. Keys must be unique across objects in a Result. 169 170 :param obj_list: an optional list of objects to send to Aria Operations. 171 Objects can be added later using add_object 172 173 :param target_definition: an optional description of the returned objects, 174 used for validation purposes 175 """ 176 self.objects: dict[Key, Object] = {} 177 if type(obj_list) is list: 178 self.add_objects(obj_list) 179 self.definition: AdapterDefinition = target_definition 180 self.adapter_type = None 181 if self.definition: 182 self.adapter_type = self.definition.key 183 self._error_message: Optional[str] = None 184 self.update_relationships: RelationshipUpdateMode = RelationshipUpdateModes.AUTO
Initializes a Result
A result contains objects, which can be added at initialization or later. Each object has a key containing one or more identifiers plus the object type and adapter type. Keys must be unique across objects in a Result.
Parameters
obj_list: an optional list of objects to send to Aria Operations. Objects can be added later using add_object
target_definition: an optional description of the returned objects, used for validation purposes
189 def object( 190 self, 191 adapter_kind: str, 192 object_kind: str, 193 name: str, 194 identifiers: Optional[list[Identifier]] = None, 195 ) -> Object: 196 """Get or create the object with key specified by adapter_kind, object_kind, 197 name, and identifiers. 198 199 This is the preferred method for creating new Objects. If this method is used 200 exclusively, all object references with the same key will point to the same 201 object. 202 203 If an object with the same key already exists in the result, return that 204 object, otherwise create a new object, add it to the result, and return it. 205 See discussion on keys in the documentation for the :class:`object.Key` class. 206 207 If this method is used to create an object, it does not need to be added 208 later using `add_object` (or `add_objects`) 209 210 :param adapter_kind: The adapter kind of the object 211 :param object_kind: The resource kind of the object 212 :param name: The name of the object 213 :param identifiers: 214 :return: The object with the given key 215 """ 216 obj = Object(Key(adapter_kind, object_kind, name, identifiers)) 217 return self.objects.setdefault(obj.get_key(), obj)
Get or create the object with key specified by adapter_kind, object_kind, name, and identifiers.
This is the preferred method for creating new Objects. If this method is used exclusively, all object references with the same key will point to the same object.
If an object with the same key already exists in the result, return that
object, otherwise create a new object, add it to the result, and return it.
See discussion on keys in the documentation for the object.Key
class.
If this method is used to create an object, it does not need to be added
later using add_object
(or add_objects
)
Parameters
- adapter_kind: The adapter kind of the object
- object_kind: The resource kind of the object
- name: The name of the object
- identifiers:
Returns
The object with the given key
219 def get_object(self, obj_key: Key) -> Optional[Object]: 220 """Get and return the object corresponding to the given key, if it exists 221 222 :param obj_key: The object key to search for 223 :return: The object with the given key, or None if the key is not in the result 224 """ 225 return self.objects.get(obj_key, None)
Get and return the object corresponding to the given key, if it exists
Parameters
- obj_key: The object key to search for
Returns
The object with the given key, or None if the key is not in the result
227 def get_objects_by_type( 228 self, object_type: str, adapter_type: Optional[str] = None 229 ) -> List[Object]: 230 """Returns all objects with the given type. If adapter_type is present, 231 the objects must also be from the given adapter type. 232 233 :param object_type: The object type to return 234 :param adapter_type: The adapter type of the objects to return 235 :return: A list of objects matching the object type and adapter type 236 """ 237 return [ 238 obj 239 for obj in self.objects.values() 240 if obj.adapter_type() == object_type 241 and (adapter_type is None or adapter_type == obj.adapter_type()) 242 ]
Returns all objects with the given type. If adapter_type is present, the objects must also be from the given adapter type.
Parameters
- object_type: The object type to return
- adapter_type: The adapter type of the objects to return
Returns
A list of objects matching the object type and adapter type
244 def add_object(self, obj: Object) -> Object: 245 """Adds the given object to the Result and returns it. 246 247 Adds the given object to the Result and returns it. A different object with 248 the same key cannot already exist in the Result. If it does, 249 an :class:`ObjectKeyAlreadyExistsException` will be raised. 250 251 :param obj: An object to add to the Result 252 :return: The object 253 :raises: ObjectKeyAlreadyExistsException if a different object with the same key 254 already exists in the Result 255 """ 256 o = self.objects.setdefault(obj.get_key(), obj) 257 if o is obj: 258 return o 259 raise ObjectKeyAlreadyExistsException( 260 f"A different object with key {obj.get_key()} already exists." 261 )
Adds the given object to the Result and returns it.
Adds the given object to the Result and returns it. A different object with
the same key cannot already exist in the Result. If it does,
an ObjectKeyAlreadyExistsException
will be raised.
Parameters
- obj: An object to add to the Result
Returns
The object
Raises
- ObjectKeyAlreadyExistsException if a different object with the same key already exists in the Result
263 def add_objects(self, obj_list: list[Object]) -> None: 264 """Adds the given objects to the Result and returns it. 265 266 Adds the given objects to the Result. A different object with the same key 267 cannot already exist in the Result. If it does, an 268 :class:`ObjectKeyAlreadyExistsException` will be raised. 269 270 :param obj_list: A list of objects to add to the Result 271 :return: The object 272 :raises: ObjectKeyAlreadyExistsException if a different object with the same 273 key already exists in the Result 274 """ 275 for obj in obj_list: 276 self.add_object(obj)
Adds the given objects to the Result and returns it.
Adds the given objects to the Result. A different object with the same key
cannot already exist in the Result. If it does, an
ObjectKeyAlreadyExistsException
will be raised.
Parameters
- obj_list: A list of objects to add to the Result
Returns
The object
Raises
- ObjectKeyAlreadyExistsException if a different object with the same key already exists in the Result
278 def with_error(self, error_message: str) -> None: 279 """Set the Adapter Instance to an error state with the provided message. 280 281 If this method is called multiple times, only the most recent error message 282 will be recorded. If error_message is set, no results (objects, relationships) 283 will be returned. 284 285 :param error_message: A string containing the error message 286 :return: None 287 """ 288 self._error_message = error_message
Set the Adapter Instance to an error state with the provided message.
If this method is called multiple times, only the most recent error message will be recorded. If error_message is set, no results (objects, relationships) will be returned.
Parameters
- error_message: A string containing the error message
Returns
None
290 def get_json(self) -> dict: 291 """Get a JSON representation of this Result 292 293 Returns a JSON representation of this Result in the format required by Aria 294 Operations. The representation includes all objects (including the object's 295 events, properties, and metrics) in the Result. Relationships are returned 296 following the update_relationships flag (See RelationshipUpdateMode). 297 298 :return: A JSON representation of this Result 299 """ 300 if self._error_message is None: 301 result = { 302 "result": [ 303 obj.get_json() 304 for obj in self.objects.values() 305 if not self._object_is_external(obj) or obj.has_content() 306 ], 307 "relationships": [], 308 "nonExistingObjects": [], 309 } 310 if ( 311 self.update_relationships == RelationshipUpdateModes.ALL 312 or self.update_relationships == RelationshipUpdateModes.PER_OBJECT 313 or ( 314 self.update_relationships == RelationshipUpdateModes.AUTO 315 and any([obj._updated_children for obj in self.objects.values()]) 316 ) 317 ): 318 result.update( 319 { 320 "relationships": [ 321 { 322 "parent": obj.get_key().get_json(), 323 "children": [ 324 child_key.get_json() 325 for child_key in obj.get_children() 326 ], 327 } 328 for obj in self.objects.values() 329 if ( 330 self.update_relationships 331 == RelationshipUpdateModes.PER_OBJECT 332 and obj._updated_children 333 ) 334 or not self.update_relationships 335 == RelationshipUpdateModes.PER_OBJECT 336 ], 337 } 338 ) 339 return result 340 else: 341 return {"errorMessage": self._error_message}
Get a JSON representation of this Result
Returns a JSON representation of this Result in the format required by Aria Operations. The representation includes all objects (including the object's events, properties, and metrics) in the Result. Relationships are returned following the update_relationships flag (See RelationshipUpdateMode).
Returns
A JSON representation of this Result
343 def send_results(self, output_pipe: str = sys.argv[-1]) -> None: 344 """Opens the output pipe and sends results directly back to the server 345 346 This method can only be called once per collection. 347 """ 348 # The server always invokes methods with the output file as the last argument 349 write_to_pipe(output_pipe, self.get_json())
Opens the output pipe and sends results directly back to the server
This method can only be called once per collection.