aria.ops.suite_api_client

  1#  Copyright 2022 VMware, Inc.
  2#  SPDX-License-Identifier: Apache-2.0
  3from __future__ import annotations
  4
  5import json
  6import logging
  7import math
  8from types import TracebackType
  9from typing import Any
 10from typing import Callable
 11from typing import Dict
 12from typing import Optional
 13from typing import Type
 14
 15import requests
 16import urllib3
 17from aria.ops.object import Identifier
 18from aria.ops.object import Key
 19from aria.ops.object import Object
 20from requests import Response
 21
 22urllib3.disable_warnings(urllib3.exceptions.InsecureRequestWarning)
 23
 24logger = logging.getLogger(__name__)
 25
 26
 27class SuiteApiConnectionParameters:
 28    def __init__(
 29        self, host: str, username: str, password: str, auth_source: str = "LOCAL"
 30    ):
 31        """Initialize SuiteApi Connection Parameters
 32
 33        :param host: The host to use for connecting to the SuiteAPI.
 34        :param username: Username used to authenticate to SuiteAPI
 35        :param password: Password used to authenticate to SuiteAPI
 36        :param auth_source: Source of authentication
 37        """
 38        if "http" in host:
 39            self.host = f"{host}/suite-api/"
 40        else:
 41            self.host = f"https://{host}/suite-api/"
 42        self.username = username
 43        self.password = password
 44        self.auth_source = auth_source
 45
 46
 47class SuiteApiClient:
 48    """Class for simplifying calls to the SuiteAPI
 49
 50    Automatically handles:
 51    * Token based authentication
 52    * Required headers
 53    * Releasing tokens (when used in a 'with' statement)
 54    * Paging (when using 'paged_get' or 'paged_post')
 55    * Logging requests
 56
 57    This class is intended to be used in a with statement:
 58    with VROpsSuiteAPIClient() as suiteApiClient:
 59        # Code using suiteApiClient goes here
 60        ...
 61    """
 62
 63    def __init__(self, connection_params: SuiteApiConnectionParameters):
 64        """Initializes a SuiteAPI client.
 65
 66        :param connection_params: Connection parameters for the Suite API.
 67        """
 68        self.credential = connection_params
 69        self.token = ""
 70
 71    def __enter__(self) -> SuiteApiClient:
 72        """Acquire a token upon entering the 'with' context
 73
 74        :return: self
 75        """
 76        self.token = self.get_token()
 77        return self
 78
 79    def __exit__(
 80        self,
 81        exception_type: Optional[Type[BaseException]],
 82        exception_value: Optional[BaseException],
 83        traceback: Optional[TracebackType],
 84    ) -> None:
 85        """Release the token upon exiting the 'with' context
 86
 87        :param exception_type: Unused
 88        :param exception_value: Unused
 89        :param traceback: Unused
 90        :return: None
 91        """
 92        self.release_token()
 93
 94    def get_token(self) -> str:
 95        """Get the authentication token
 96
 97        Gets the current authentication token. If no current token exists, acquires an authentication token first.
 98
 99        :return: The authentication token
100        """
101        if self.token == "":
102            with self.post(
103                "/api/auth/token/acquire",
104                json={
105                    "username": self.credential.username,
106                    "password": self.credential.password,
107                    "authSource": self.credential.auth_source,
108                },
109            ) as token_response:
110                if token_response.ok:
111                    self.token = token_response.json()["token"]
112                    logger.debug("Acquired token " + self.token)
113                else:
114                    logger.warning(
115                        f"Could not acquire SuiteAPI token: {token_response}"
116                    )
117
118        return self.token
119
120    def release_token(self) -> None:
121        """Release the authentication token, if it exists
122
123        :return: None
124        """
125        if self.token != "":
126            self.post("auth/token/release").close()
127            self.token = ""
128
129    def get(self, url: str, **kwargs: Any) -> Response:
130        """Send a GET request to the SuiteAPI
131        The 'Response' object should be used in a 'with' block or
132        manually closed after use
133
134        :param url: URL to send GET request to
135        :param kwargs: Additional keyword arguments to pass to request
136        :return: The API response
137        """
138        return self._request_wrapper(requests.get, url, **kwargs)
139
140    def paged_get(self, url: str, key: str, **kwargs: Any) -> dict:
141        """Send a GET request to the SuiteAPI that gets a paged response
142
143        :param url: URL to send GET request to
144        :param key: Json key that contains the paged data
145        :param kwargs: Additional keyword arguments to pass to request
146        :return: The API response
147        """
148        return self._paged_request(requests.get, url, key, **kwargs)
149
150    def post(self, url: str, **kwargs: Any) -> Response:
151        """Send a POST request to the SuiteAPI
152        The 'Response' object should be used in a 'with' block or
153        manually closed after use
154
155        :param url: URL to send POST request to
156        :param kwargs: Additional keyword arguments to pass to request
157        :return: The API response
158        """
159        kwargs.setdefault("headers", {})
160        kwargs["headers"].setdefault("Content-Type", "application/json")
161        return self._request_wrapper(requests.post, url, **kwargs)
162
163    def paged_post(self, url: str, key: str, **kwargs: Any) -> dict:
164        """Send a POST request to the SuiteAPI that gets a paged response.
165
166        :param url: URL to send POST request to
167        :param key: Json key that contains the paged data
168        :param kwargs: Additional keyword arguments to pass to request
169        :return: The API response
170        """
171        kwargs.setdefault("headers", {})
172        kwargs["headers"].setdefault("Content-Type", "application/json")
173        return self._paged_request(requests.post, url, key, **kwargs)
174
175    def put(self, url: str, **kwargs: Any) -> Response:
176        """Send a PUT request to the SuiteAPI
177        The 'Response' object should be used in a 'with' block or
178        manually closed after use
179
180        :param url: URL to send PUT request to
181        :param kwargs: Additional keyword arguments to pass to request
182        :return: The API response
183        """
184        return self._request_wrapper(requests.put, url, **kwargs)
185
186    def patch(self, url: str, **kwargs: Any) -> Response:
187        """Send a PATCH request to the SuiteAPI
188        The 'Response' object should be used in a 'with' block or
189        manually closed after use
190
191        :param url: URL to send PATCH request to
192        :param kwargs: Additional keyword arguments to pass to request
193        :return: The API response
194        """
195        return self._request_wrapper(requests.patch, url, **kwargs)
196
197    def delete(self, url: str, **kwargs: Any) -> Response:
198        """Send a DELETE request to the SuiteAPI
199        The 'Response' object should be used in a 'with' block or
200        manually closed after use
201
202        :param url: URL to send DELETE request to
203        :param kwargs: Additional keyword arguments to pass to request
204        :return: The API response
205        """
206        return self._request_wrapper(requests.delete, url, **kwargs)
207
208    def _add_paging(self, **kwargs: Any) -> dict:
209        kwargs.setdefault("params", {})
210        kwargs["params"].setdefault("page", 0)
211        kwargs["params"].setdefault("pageSize", 1000)
212
213        if "page" in kwargs:
214            kwargs["params"]["page"] = kwargs.pop("page")
215        if "pageSize" in kwargs:
216            kwargs["params"]["pageSize"] = kwargs.pop("pageSize")
217
218        return kwargs
219
220    # Implementations for common endpoints:
221
222    def query_for_resources(self, query: Dict[str, Any]) -> list[Object]:
223        """Query for resources using the Suite API, and convert the
224        responses to SDK Objects.
225
226        Note that not all information from the query is returned. For example, the
227        query returns health statuses of each object, but those are not present in
228        the resulting Objects. If information other than the Object itself is needed,
229        you will need to call the endpoint and process the results manually.
230
231        :param query: json of the resourceQuery, as defined in the SuiteAPI docs:
232        https://[[aria-ops-hostname]]/suite-api/doc/swagger-ui.html#/Resources/getMatchingResourcesUsingPOST
233        :return list of sdk Objects representing each of the returned objects.
234        """
235        try:
236            results = []
237            if "name" in query and "regex" in query:
238                # This is behavior in the suite api itself, we're just warning about it
239                # here to avoid confusion.
240                logger.warning(
241                    "'name' and 'regex' are mutually exclusive in resource "
242                    "queries. Ignoring the 'regex' key in favor of 'name' "
243                    "key."
244                )
245            # The 'name' key takes an array but only looks up the first element.
246            # Fix that limitation here.
247            if "name" in query and len(query["name"]) > 1:
248                json_body = query.copy()
249                # TODO: Improve concurrancy when we add async support
250                #  to suite_api_client
251                for name in query["name"]:
252                    json_body.update({"name": [name]})
253                    response = self.paged_post(
254                        "/api/resources/query", "resourceList", json=json_body
255                    )
256                    results.extend(response.get("resourceList", []))
257            else:
258                response = self.paged_post(
259                    "/api/resources/query",
260                    "resourceList",
261                    json=query,
262                )
263                results = response.get("resourceList", [])
264            return [key_to_object(obj["resourceKey"]) for obj in results]
265        except Exception as e:
266            logger.error(e)
267            logger.exception(e)
268            return []
269
270    def _paged_request(
271        self, request_func: Callable, url: str, key: str, **kwargs: Any
272    ) -> dict:
273        """Send a request to the SuiteAPI that returns a paged response. Each response must have data returned in an
274        array at key 'key'. The array from the responses will be combined into a single array and returned in a map of
275        the form:
276        {
277           "{key}": [aggregated data]
278        }
279
280        :param url: URL to send request to
281        :param key: Json key that contains the paged data
282        :param kwargs: Additional keyword arguments to pass to request
283        :return: The API response
284        """
285        kwargs = self._add_paging(**kwargs)
286        with self._request_wrapper(request_func, url, **kwargs) as page_0:
287            if page_0.status_code < 300:
288                page_0_body = json.loads(page_0.text)
289            else:
290                # _request_wrapper will log the error
291                # TODO: How should we communicate to caller that
292                #       request(s) have failed?
293                return {key: []}
294        total_objects = int(
295            page_0_body.get("pageInfo", {"totalCount": 1}).get("totalCount", 1)
296        )
297        page_size = kwargs["params"]["pageSize"]
298        remaining_pages = math.ceil(total_objects / page_size) - 1
299        objects = page_0_body.get(key, [])
300        while remaining_pages > 0:
301            kwargs = self._add_paging(page=remaining_pages, **kwargs)
302            with self._request_wrapper(request_func, url, **kwargs) as page_n:
303                if page_n.status_code < 300:
304                    page_n_body = json.loads(page_n.text)
305            objects.extend(page_n_body.get(key, []))
306            remaining_pages -= 1
307        return {key: objects}
308
309    def _request_wrapper(
310        self, request_func: Callable[..., Response], url: str, **kwargs: Any
311    ) -> Response:
312        kwargs = self._to_vrops_request(url, **kwargs)
313        result = request_func(**kwargs)
314        if result.ok:
315            logger.info(
316                f"{request_func.__name__} {kwargs['url']}: OK({result.status_code})"
317            )
318        else:
319            logger.warning(
320                f"{request_func.__name__} {kwargs['url']}: ERROR({result.status_code})"
321            )
322        logger.debug(result.text)
323        return result
324
325    def _to_vrops_request(self, url: str, **kwargs: Any) -> dict:
326        kwargs.setdefault("url", url)
327        kwargs.setdefault("headers", {})
328        if self.token:
329            kwargs["headers"]["Authorization"] = "vRealizeOpsToken " + self.token
330        kwargs["headers"].setdefault("Accept", "application/json")
331        kwargs.setdefault("verify", False)
332
333        url = kwargs["url"]
334        if "internal/" in url:
335            kwargs["headers"]["X-vRealizeOps-API-use-unsupported"] = "true"
336            logger.info(f"Using unsupported API: {url}")
337        if url.startswith("http"):
338            return kwargs
339
340        if url.startswith("/"):
341            url = url[1:]
342        if url.startswith("suite-api/"):
343            url = url[10:]
344        elif url.startswith("api") or url.startswith("internal"):
345            kwargs["url"] = self.credential.host + url
346        else:
347            kwargs["url"] = self.credential.host + "api/" + url
348        return kwargs
349
350
351# Helper methods:
352
353
354def key_to_object(json_object_key: Dict[str, Any]) -> Object:
355    return Object(
356        Key(
357            json_object_key["adapterKindKey"],
358            json_object_key["resourceKindKey"],
359            json_object_key["name"],
360            [
361                Identifier(
362                    identifier["identifierType"]["name"],
363                    identifier["value"],
364                    identifier["identifierType"]["isPartOfUniqueness"],
365                )
366                for identifier in json_object_key["resourceIdentifiers"]
367            ],
368        )
369    )
class SuiteApiConnectionParameters:
28class SuiteApiConnectionParameters:
29    def __init__(
30        self, host: str, username: str, password: str, auth_source: str = "LOCAL"
31    ):
32        """Initialize SuiteApi Connection Parameters
33
34        :param host: The host to use for connecting to the SuiteAPI.
35        :param username: Username used to authenticate to SuiteAPI
36        :param password: Password used to authenticate to SuiteAPI
37        :param auth_source: Source of authentication
38        """
39        if "http" in host:
40            self.host = f"{host}/suite-api/"
41        else:
42            self.host = f"https://{host}/suite-api/"
43        self.username = username
44        self.password = password
45        self.auth_source = auth_source
SuiteApiConnectionParameters(host: str, username: str, password: str, auth_source: str = 'LOCAL')
29    def __init__(
30        self, host: str, username: str, password: str, auth_source: str = "LOCAL"
31    ):
32        """Initialize SuiteApi Connection Parameters
33
34        :param host: The host to use for connecting to the SuiteAPI.
35        :param username: Username used to authenticate to SuiteAPI
36        :param password: Password used to authenticate to SuiteAPI
37        :param auth_source: Source of authentication
38        """
39        if "http" in host:
40            self.host = f"{host}/suite-api/"
41        else:
42            self.host = f"https://{host}/suite-api/"
43        self.username = username
44        self.password = password
45        self.auth_source = auth_source

Initialize SuiteApi Connection Parameters

Parameters
  • host: The host to use for connecting to the SuiteAPI.
  • username: Username used to authenticate to SuiteAPI
  • password: Password used to authenticate to SuiteAPI
  • auth_source: Source of authentication
class SuiteApiClient:
 48class SuiteApiClient:
 49    """Class for simplifying calls to the SuiteAPI
 50
 51    Automatically handles:
 52    * Token based authentication
 53    * Required headers
 54    * Releasing tokens (when used in a 'with' statement)
 55    * Paging (when using 'paged_get' or 'paged_post')
 56    * Logging requests
 57
 58    This class is intended to be used in a with statement:
 59    with VROpsSuiteAPIClient() as suiteApiClient:
 60        # Code using suiteApiClient goes here
 61        ...
 62    """
 63
 64    def __init__(self, connection_params: SuiteApiConnectionParameters):
 65        """Initializes a SuiteAPI client.
 66
 67        :param connection_params: Connection parameters for the Suite API.
 68        """
 69        self.credential = connection_params
 70        self.token = ""
 71
 72    def __enter__(self) -> SuiteApiClient:
 73        """Acquire a token upon entering the 'with' context
 74
 75        :return: self
 76        """
 77        self.token = self.get_token()
 78        return self
 79
 80    def __exit__(
 81        self,
 82        exception_type: Optional[Type[BaseException]],
 83        exception_value: Optional[BaseException],
 84        traceback: Optional[TracebackType],
 85    ) -> None:
 86        """Release the token upon exiting the 'with' context
 87
 88        :param exception_type: Unused
 89        :param exception_value: Unused
 90        :param traceback: Unused
 91        :return: None
 92        """
 93        self.release_token()
 94
 95    def get_token(self) -> str:
 96        """Get the authentication token
 97
 98        Gets the current authentication token. If no current token exists, acquires an authentication token first.
 99
100        :return: The authentication token
101        """
102        if self.token == "":
103            with self.post(
104                "/api/auth/token/acquire",
105                json={
106                    "username": self.credential.username,
107                    "password": self.credential.password,
108                    "authSource": self.credential.auth_source,
109                },
110            ) as token_response:
111                if token_response.ok:
112                    self.token = token_response.json()["token"]
113                    logger.debug("Acquired token " + self.token)
114                else:
115                    logger.warning(
116                        f"Could not acquire SuiteAPI token: {token_response}"
117                    )
118
119        return self.token
120
121    def release_token(self) -> None:
122        """Release the authentication token, if it exists
123
124        :return: None
125        """
126        if self.token != "":
127            self.post("auth/token/release").close()
128            self.token = ""
129
130    def get(self, url: str, **kwargs: Any) -> Response:
131        """Send a GET request to the SuiteAPI
132        The 'Response' object should be used in a 'with' block or
133        manually closed after use
134
135        :param url: URL to send GET request to
136        :param kwargs: Additional keyword arguments to pass to request
137        :return: The API response
138        """
139        return self._request_wrapper(requests.get, url, **kwargs)
140
141    def paged_get(self, url: str, key: str, **kwargs: Any) -> dict:
142        """Send a GET request to the SuiteAPI that gets a paged response
143
144        :param url: URL to send GET request to
145        :param key: Json key that contains the paged data
146        :param kwargs: Additional keyword arguments to pass to request
147        :return: The API response
148        """
149        return self._paged_request(requests.get, url, key, **kwargs)
150
151    def post(self, url: str, **kwargs: Any) -> Response:
152        """Send a POST request to the SuiteAPI
153        The 'Response' object should be used in a 'with' block or
154        manually closed after use
155
156        :param url: URL to send POST request to
157        :param kwargs: Additional keyword arguments to pass to request
158        :return: The API response
159        """
160        kwargs.setdefault("headers", {})
161        kwargs["headers"].setdefault("Content-Type", "application/json")
162        return self._request_wrapper(requests.post, url, **kwargs)
163
164    def paged_post(self, url: str, key: str, **kwargs: Any) -> dict:
165        """Send a POST request to the SuiteAPI that gets a paged response.
166
167        :param url: URL to send POST request to
168        :param key: Json key that contains the paged data
169        :param kwargs: Additional keyword arguments to pass to request
170        :return: The API response
171        """
172        kwargs.setdefault("headers", {})
173        kwargs["headers"].setdefault("Content-Type", "application/json")
174        return self._paged_request(requests.post, url, key, **kwargs)
175
176    def put(self, url: str, **kwargs: Any) -> Response:
177        """Send a PUT request to the SuiteAPI
178        The 'Response' object should be used in a 'with' block or
179        manually closed after use
180
181        :param url: URL to send PUT request to
182        :param kwargs: Additional keyword arguments to pass to request
183        :return: The API response
184        """
185        return self._request_wrapper(requests.put, url, **kwargs)
186
187    def patch(self, url: str, **kwargs: Any) -> Response:
188        """Send a PATCH request to the SuiteAPI
189        The 'Response' object should be used in a 'with' block or
190        manually closed after use
191
192        :param url: URL to send PATCH request to
193        :param kwargs: Additional keyword arguments to pass to request
194        :return: The API response
195        """
196        return self._request_wrapper(requests.patch, url, **kwargs)
197
198    def delete(self, url: str, **kwargs: Any) -> Response:
199        """Send a DELETE request to the SuiteAPI
200        The 'Response' object should be used in a 'with' block or
201        manually closed after use
202
203        :param url: URL to send DELETE request to
204        :param kwargs: Additional keyword arguments to pass to request
205        :return: The API response
206        """
207        return self._request_wrapper(requests.delete, url, **kwargs)
208
209    def _add_paging(self, **kwargs: Any) -> dict:
210        kwargs.setdefault("params", {})
211        kwargs["params"].setdefault("page", 0)
212        kwargs["params"].setdefault("pageSize", 1000)
213
214        if "page" in kwargs:
215            kwargs["params"]["page"] = kwargs.pop("page")
216        if "pageSize" in kwargs:
217            kwargs["params"]["pageSize"] = kwargs.pop("pageSize")
218
219        return kwargs
220
221    # Implementations for common endpoints:
222
223    def query_for_resources(self, query: Dict[str, Any]) -> list[Object]:
224        """Query for resources using the Suite API, and convert the
225        responses to SDK Objects.
226
227        Note that not all information from the query is returned. For example, the
228        query returns health statuses of each object, but those are not present in
229        the resulting Objects. If information other than the Object itself is needed,
230        you will need to call the endpoint and process the results manually.
231
232        :param query: json of the resourceQuery, as defined in the SuiteAPI docs:
233        https://[[aria-ops-hostname]]/suite-api/doc/swagger-ui.html#/Resources/getMatchingResourcesUsingPOST
234        :return list of sdk Objects representing each of the returned objects.
235        """
236        try:
237            results = []
238            if "name" in query and "regex" in query:
239                # This is behavior in the suite api itself, we're just warning about it
240                # here to avoid confusion.
241                logger.warning(
242                    "'name' and 'regex' are mutually exclusive in resource "
243                    "queries. Ignoring the 'regex' key in favor of 'name' "
244                    "key."
245                )
246            # The 'name' key takes an array but only looks up the first element.
247            # Fix that limitation here.
248            if "name" in query and len(query["name"]) > 1:
249                json_body = query.copy()
250                # TODO: Improve concurrancy when we add async support
251                #  to suite_api_client
252                for name in query["name"]:
253                    json_body.update({"name": [name]})
254                    response = self.paged_post(
255                        "/api/resources/query", "resourceList", json=json_body
256                    )
257                    results.extend(response.get("resourceList", []))
258            else:
259                response = self.paged_post(
260                    "/api/resources/query",
261                    "resourceList",
262                    json=query,
263                )
264                results = response.get("resourceList", [])
265            return [key_to_object(obj["resourceKey"]) for obj in results]
266        except Exception as e:
267            logger.error(e)
268            logger.exception(e)
269            return []
270
271    def _paged_request(
272        self, request_func: Callable, url: str, key: str, **kwargs: Any
273    ) -> dict:
274        """Send a request to the SuiteAPI that returns a paged response. Each response must have data returned in an
275        array at key 'key'. The array from the responses will be combined into a single array and returned in a map of
276        the form:
277        {
278           "{key}": [aggregated data]
279        }
280
281        :param url: URL to send request to
282        :param key: Json key that contains the paged data
283        :param kwargs: Additional keyword arguments to pass to request
284        :return: The API response
285        """
286        kwargs = self._add_paging(**kwargs)
287        with self._request_wrapper(request_func, url, **kwargs) as page_0:
288            if page_0.status_code < 300:
289                page_0_body = json.loads(page_0.text)
290            else:
291                # _request_wrapper will log the error
292                # TODO: How should we communicate to caller that
293                #       request(s) have failed?
294                return {key: []}
295        total_objects = int(
296            page_0_body.get("pageInfo", {"totalCount": 1}).get("totalCount", 1)
297        )
298        page_size = kwargs["params"]["pageSize"]
299        remaining_pages = math.ceil(total_objects / page_size) - 1
300        objects = page_0_body.get(key, [])
301        while remaining_pages > 0:
302            kwargs = self._add_paging(page=remaining_pages, **kwargs)
303            with self._request_wrapper(request_func, url, **kwargs) as page_n:
304                if page_n.status_code < 300:
305                    page_n_body = json.loads(page_n.text)
306            objects.extend(page_n_body.get(key, []))
307            remaining_pages -= 1
308        return {key: objects}
309
310    def _request_wrapper(
311        self, request_func: Callable[..., Response], url: str, **kwargs: Any
312    ) -> Response:
313        kwargs = self._to_vrops_request(url, **kwargs)
314        result = request_func(**kwargs)
315        if result.ok:
316            logger.info(
317                f"{request_func.__name__} {kwargs['url']}: OK({result.status_code})"
318            )
319        else:
320            logger.warning(
321                f"{request_func.__name__} {kwargs['url']}: ERROR({result.status_code})"
322            )
323        logger.debug(result.text)
324        return result
325
326    def _to_vrops_request(self, url: str, **kwargs: Any) -> dict:
327        kwargs.setdefault("url", url)
328        kwargs.setdefault("headers", {})
329        if self.token:
330            kwargs["headers"]["Authorization"] = "vRealizeOpsToken " + self.token
331        kwargs["headers"].setdefault("Accept", "application/json")
332        kwargs.setdefault("verify", False)
333
334        url = kwargs["url"]
335        if "internal/" in url:
336            kwargs["headers"]["X-vRealizeOps-API-use-unsupported"] = "true"
337            logger.info(f"Using unsupported API: {url}")
338        if url.startswith("http"):
339            return kwargs
340
341        if url.startswith("/"):
342            url = url[1:]
343        if url.startswith("suite-api/"):
344            url = url[10:]
345        elif url.startswith("api") or url.startswith("internal"):
346            kwargs["url"] = self.credential.host + url
347        else:
348            kwargs["url"] = self.credential.host + "api/" + url
349        return kwargs

Class for simplifying calls to the SuiteAPI

Automatically handles:

  • Token based authentication
  • Required headers
  • Releasing tokens (when used in a 'with' statement)
  • Paging (when using 'paged_get' or 'paged_post')
  • Logging requests

This class is intended to be used in a with statement: with VROpsSuiteAPIClient() as suiteApiClient: # Code using suiteApiClient goes here ...

SuiteApiClient( connection_params: aria.ops.suite_api_client.SuiteApiConnectionParameters)
64    def __init__(self, connection_params: SuiteApiConnectionParameters):
65        """Initializes a SuiteAPI client.
66
67        :param connection_params: Connection parameters for the Suite API.
68        """
69        self.credential = connection_params
70        self.token = ""

Initializes a SuiteAPI client.

Parameters
  • connection_params: Connection parameters for the Suite API.
def get_token(self) -> str:
 95    def get_token(self) -> str:
 96        """Get the authentication token
 97
 98        Gets the current authentication token. If no current token exists, acquires an authentication token first.
 99
100        :return: The authentication token
101        """
102        if self.token == "":
103            with self.post(
104                "/api/auth/token/acquire",
105                json={
106                    "username": self.credential.username,
107                    "password": self.credential.password,
108                    "authSource": self.credential.auth_source,
109                },
110            ) as token_response:
111                if token_response.ok:
112                    self.token = token_response.json()["token"]
113                    logger.debug("Acquired token " + self.token)
114                else:
115                    logger.warning(
116                        f"Could not acquire SuiteAPI token: {token_response}"
117                    )
118
119        return self.token

Get the authentication token

Gets the current authentication token. If no current token exists, acquires an authentication token first.

Returns

The authentication token

def release_token(self) -> None:
121    def release_token(self) -> None:
122        """Release the authentication token, if it exists
123
124        :return: None
125        """
126        if self.token != "":
127            self.post("auth/token/release").close()
128            self.token = ""

Release the authentication token, if it exists

Returns

None

def get(self, url: str, **kwargs: Any) -> requests.models.Response:
130    def get(self, url: str, **kwargs: Any) -> Response:
131        """Send a GET request to the SuiteAPI
132        The 'Response' object should be used in a 'with' block or
133        manually closed after use
134
135        :param url: URL to send GET request to
136        :param kwargs: Additional keyword arguments to pass to request
137        :return: The API response
138        """
139        return self._request_wrapper(requests.get, url, **kwargs)

Send a GET request to the SuiteAPI The 'Response' object should be used in a 'with' block or manually closed after use

Parameters
  • url: URL to send GET request to
  • kwargs: Additional keyword arguments to pass to request
Returns

The API response

def paged_get(self, url: str, key: str, **kwargs: Any) -> dict:
141    def paged_get(self, url: str, key: str, **kwargs: Any) -> dict:
142        """Send a GET request to the SuiteAPI that gets a paged response
143
144        :param url: URL to send GET request to
145        :param key: Json key that contains the paged data
146        :param kwargs: Additional keyword arguments to pass to request
147        :return: The API response
148        """
149        return self._paged_request(requests.get, url, key, **kwargs)

Send a GET request to the SuiteAPI that gets a paged response

Parameters
  • url: URL to send GET request to
  • key: Json key that contains the paged data
  • kwargs: Additional keyword arguments to pass to request
Returns

The API response

def post(self, url: str, **kwargs: Any) -> requests.models.Response:
151    def post(self, url: str, **kwargs: Any) -> Response:
152        """Send a POST request to the SuiteAPI
153        The 'Response' object should be used in a 'with' block or
154        manually closed after use
155
156        :param url: URL to send POST request to
157        :param kwargs: Additional keyword arguments to pass to request
158        :return: The API response
159        """
160        kwargs.setdefault("headers", {})
161        kwargs["headers"].setdefault("Content-Type", "application/json")
162        return self._request_wrapper(requests.post, url, **kwargs)

Send a POST request to the SuiteAPI The 'Response' object should be used in a 'with' block or manually closed after use

Parameters
  • url: URL to send POST request to
  • kwargs: Additional keyword arguments to pass to request
Returns

The API response

def paged_post(self, url: str, key: str, **kwargs: Any) -> dict:
164    def paged_post(self, url: str, key: str, **kwargs: Any) -> dict:
165        """Send a POST request to the SuiteAPI that gets a paged response.
166
167        :param url: URL to send POST request to
168        :param key: Json key that contains the paged data
169        :param kwargs: Additional keyword arguments to pass to request
170        :return: The API response
171        """
172        kwargs.setdefault("headers", {})
173        kwargs["headers"].setdefault("Content-Type", "application/json")
174        return self._paged_request(requests.post, url, key, **kwargs)

Send a POST request to the SuiteAPI that gets a paged response.

Parameters
  • url: URL to send POST request to
  • key: Json key that contains the paged data
  • kwargs: Additional keyword arguments to pass to request
Returns

The API response

def put(self, url: str, **kwargs: Any) -> requests.models.Response:
176    def put(self, url: str, **kwargs: Any) -> Response:
177        """Send a PUT request to the SuiteAPI
178        The 'Response' object should be used in a 'with' block or
179        manually closed after use
180
181        :param url: URL to send PUT request to
182        :param kwargs: Additional keyword arguments to pass to request
183        :return: The API response
184        """
185        return self._request_wrapper(requests.put, url, **kwargs)

Send a PUT request to the SuiteAPI The 'Response' object should be used in a 'with' block or manually closed after use

Parameters
  • url: URL to send PUT request to
  • kwargs: Additional keyword arguments to pass to request
Returns

The API response

def patch(self, url: str, **kwargs: Any) -> requests.models.Response:
187    def patch(self, url: str, **kwargs: Any) -> Response:
188        """Send a PATCH request to the SuiteAPI
189        The 'Response' object should be used in a 'with' block or
190        manually closed after use
191
192        :param url: URL to send PATCH request to
193        :param kwargs: Additional keyword arguments to pass to request
194        :return: The API response
195        """
196        return self._request_wrapper(requests.patch, url, **kwargs)

Send a PATCH request to the SuiteAPI The 'Response' object should be used in a 'with' block or manually closed after use

Parameters
  • url: URL to send PATCH request to
  • kwargs: Additional keyword arguments to pass to request
Returns

The API response

def delete(self, url: str, **kwargs: Any) -> requests.models.Response:
198    def delete(self, url: str, **kwargs: Any) -> Response:
199        """Send a DELETE request to the SuiteAPI
200        The 'Response' object should be used in a 'with' block or
201        manually closed after use
202
203        :param url: URL to send DELETE request to
204        :param kwargs: Additional keyword arguments to pass to request
205        :return: The API response
206        """
207        return self._request_wrapper(requests.delete, url, **kwargs)

Send a DELETE request to the SuiteAPI The 'Response' object should be used in a 'with' block or manually closed after use

Parameters
  • url: URL to send DELETE request to
  • kwargs: Additional keyword arguments to pass to request
Returns

The API response

def query_for_resources(self, query: Dict[str, Any]) -> list[aria.ops.object.Object]:
223    def query_for_resources(self, query: Dict[str, Any]) -> list[Object]:
224        """Query for resources using the Suite API, and convert the
225        responses to SDK Objects.
226
227        Note that not all information from the query is returned. For example, the
228        query returns health statuses of each object, but those are not present in
229        the resulting Objects. If information other than the Object itself is needed,
230        you will need to call the endpoint and process the results manually.
231
232        :param query: json of the resourceQuery, as defined in the SuiteAPI docs:
233        https://[[aria-ops-hostname]]/suite-api/doc/swagger-ui.html#/Resources/getMatchingResourcesUsingPOST
234        :return list of sdk Objects representing each of the returned objects.
235        """
236        try:
237            results = []
238            if "name" in query and "regex" in query:
239                # This is behavior in the suite api itself, we're just warning about it
240                # here to avoid confusion.
241                logger.warning(
242                    "'name' and 'regex' are mutually exclusive in resource "
243                    "queries. Ignoring the 'regex' key in favor of 'name' "
244                    "key."
245                )
246            # The 'name' key takes an array but only looks up the first element.
247            # Fix that limitation here.
248            if "name" in query and len(query["name"]) > 1:
249                json_body = query.copy()
250                # TODO: Improve concurrancy when we add async support
251                #  to suite_api_client
252                for name in query["name"]:
253                    json_body.update({"name": [name]})
254                    response = self.paged_post(
255                        "/api/resources/query", "resourceList", json=json_body
256                    )
257                    results.extend(response.get("resourceList", []))
258            else:
259                response = self.paged_post(
260                    "/api/resources/query",
261                    "resourceList",
262                    json=query,
263                )
264                results = response.get("resourceList", [])
265            return [key_to_object(obj["resourceKey"]) for obj in results]
266        except Exception as e:
267            logger.error(e)
268            logger.exception(e)
269            return []

Query for resources using the Suite API, and convert the responses to SDK Objects.

Note that not all information from the query is returned. For example, the query returns health statuses of each object, but those are not present in the resulting Objects. If information other than the Object itself is needed, you will need to call the endpoint and process the results manually.

Parameters
  • query: json of the resourceQuery, as defined in the SuiteAPI docs: https://[[aria-ops-hostname]]/suite-api/doc/swagger-ui.html#/Resources/getMatchingResourcesUsingPOST :return list of sdk Objects representing each of the returned objects.
def key_to_object(json_object_key: Dict[str, Any]) -> aria.ops.object.Object:
355def key_to_object(json_object_key: Dict[str, Any]) -> Object:
356    return Object(
357        Key(
358            json_object_key["adapterKindKey"],
359            json_object_key["resourceKindKey"],
360            json_object_key["name"],
361            [
362                Identifier(
363                    identifier["identifierType"]["name"],
364                    identifier["value"],
365                    identifier["identifierType"]["isPartOfUniqueness"],
366                )
367                for identifier in json_object_key["resourceIdentifiers"]
368            ],
369        )
370    )