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 )
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
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
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 ...
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.
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
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
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
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
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
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
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
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
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
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.
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 )