|
| 1 | +from typing import Any, Callable |
1 | 2 | import requests
|
2 | 3 |
|
3 | 4 | from ._common import AppDException
|
@@ -211,3 +212,55 @@ def _full_uri(self, endpoint: str) -> str:
|
211 | 212 | str: The URI.
|
212 | 213 | """
|
213 | 214 | return f"{self.CONTROLLER_BASE_URL}{endpoint}"
|
| 215 | + |
| 216 | + @staticmethod |
| 217 | + def __request_or_raise(method: str, |
| 218 | + uri: str, |
| 219 | + object_name: str, |
| 220 | + json_decode: bool = True, |
| 221 | + expected_status_code: int = 200, |
| 222 | + **kwargs: dict[str, str]): |
| 223 | + """Private method. |
| 224 | +
|
| 225 | + Uplink-style decorator for requests. Use like `_request_or_raise` but as decorator. |
| 226 | +
|
| 227 | + URI and object name are expanded at runtime as `URITemplate`. |
| 228 | +
|
| 229 | + Example usage: |
| 230 | + ``` |
| 231 | + @__request_or_raise("GET", |
| 232 | + "/controller/rest/applications/{application_name}", |
| 233 | + "application {application_name}", |
| 234 | + headers={"myHeader": "value"}) |
| 235 | + def get_application_decorated(application_name): |
| 236 | + \"""Get application by name.\""" |
| 237 | + ``` |
| 238 | + """ |
| 239 | + |
| 240 | + from inspect import signature |
| 241 | + from uritemplate import URITemplate |
| 242 | + |
| 243 | + def _inner_request_or_raise_decorator( |
| 244 | + func: Callable[[Any], Any]) -> Callable[[Any], str | list[dict[str, str]]]: |
| 245 | + """Handles function.""" |
| 246 | + |
| 247 | + def __inner_request_or_raise_decorator(*args: list[Any]) -> str | list[dict[str, str]]: |
| 248 | + """Handles arguments passed to function.""" |
| 249 | + self: AppDController = args[0] # type: ignore |
| 250 | + args = args[1:] |
| 251 | + |
| 252 | + bound_args = signature(func).bind(*args).arguments |
| 253 | + expanded_uri = URITemplate(self._full_uri(uri)).expand(bound_args) |
| 254 | + expanded_object_name = URITemplate(object_name).expand(bound_args) |
| 255 | + |
| 256 | + k = kwargs |
| 257 | + if json_decode: |
| 258 | + k = self._safe_add_to_kwargs("params", "output", "JSON", **k) |
| 259 | + |
| 260 | + res: requests.Response = self._request_or_raise(method, expanded_uri, expanded_object_name, |
| 261 | + expected_status_code, **k) |
| 262 | + return res.json() if json_decode else res.text |
| 263 | + |
| 264 | + return __inner_request_or_raise_decorator |
| 265 | + |
| 266 | + return _inner_request_or_raise_decorator |
0 commit comments