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())
class ObjectKeyAlreadyExistsException(builtins.Exception):
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
class TestResult:
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

TestResult()
28    def __init__(self) -> None:
29        """Initializes a Result"""
30        self._error_message: Optional[str] = None

Initializes a Result

def is_success(self) -> bool:
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

def with_error(self, error_message: str) -> None:
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

def get_json(self) -> dict:
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

def send_results(self, output_pipe: str = sys.argv[-1]) -> None:
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.

class EndpointResult:
 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.

EndpointResult()
83    def __init__(self) -> None:
84        """Initializes an EndpointResult"""
85        self.endpoints: list[str] = []

Initializes an EndpointResult

def with_endpoint(self, endpoint_url: str) -> None:
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

def get_json(self) -> dict:
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

def send_results(self, output_pipe: str = sys.argv[-1]) -> None:
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.

class RelationshipUpdateModes(aenum.Enum):
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.

class CollectResult:
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

CollectResult( obj_list: Optional[list[aria.ops.object.Object]] = None, target_definition: aria.ops.definition.adapter_definition.AdapterDefinition = None)
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

def object( self, adapter_kind: str, object_kind: str, name: str, identifiers: Optional[list[aria.ops.object.Identifier]] = None) -> aria.ops.object.Object:
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

def get_object(self, obj_key: aria.ops.object.Key) -> Optional[aria.ops.object.Object]:
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

def get_objects_by_type( self, object_type: str, adapter_type: Optional[str] = None) -> List[aria.ops.object.Object]:
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

def add_object(self, obj: aria.ops.object.Object) -> aria.ops.object.Object:
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
def add_objects(self, obj_list: list[aria.ops.object.Object]) -> None:
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
def with_error(self, error_message: str) -> None:
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

def get_json(self) -> dict:
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

def send_results(self, output_pipe: str = sys.argv[-1]) -> None:
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.