-
Notifications
You must be signed in to change notification settings - Fork 297
Commit
This commit does not belong to any branch on this repository, and may belong to a fork outside of the repository.
Rewrite LabelAPIClient to be able to return error messages from Label Repo API. Main features: 1. Raises LabelRepoAPIException when response code is 400 or 500 level. 2. Always return response as a second argument to further inspect it, if necessary.
- Loading branch information
1 parent
96b88ed
commit 8c82dac
Showing
3 changed files
with
132 additions
and
43 deletions.
There are no files selected for viewing
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -1,42 +1,117 @@ | ||
| import typing | ||
| from urllib.parse import urljoin | ||
|
|
||
| from apps.grafana_plugin.helpers.client import APIClient | ||
| import requests | ||
| from django.conf import settings | ||
|
|
||
| if typing.TYPE_CHECKING: | ||
| from apps.labels.utils import LabelKeyData, LabelsKeysData, LabelUpdateParam | ||
|
|
||
|
|
||
| class LabelsAPIClient(APIClient): | ||
| class LabelsRepoAPIException(Exception): | ||
| """A generic 400 or 500 level exception from the Label Repo API""" | ||
|
|
||
| def __init__(self, status, url, msg="", method="GET"): | ||
| self.url = url | ||
| self.status = status | ||
| self.method = method | ||
|
|
||
| # Error-message returned by label repo. | ||
| # If status is 400 level it will contain user-visible error message. | ||
| self.msg = msg | ||
|
|
||
| def __str__(self): | ||
| return f"LabelsRepoAPIException: status={self.status} url={self.url} method={self.method}" | ||
|
|
||
|
|
||
| TIMEOUT = 5 | ||
|
|
||
|
|
||
| class LabelsAPIClient: | ||
| LABELS_API_URL = "/api/plugins/grafana-labels-app/resources/v1/labels/" | ||
|
|
||
| def __init__(self, api_url: str, api_token: str) -> None: | ||
| super().__init__(api_url, api_token) | ||
| self.api_token = api_token | ||
| self.api_url = urljoin(api_url, self.LABELS_API_URL) | ||
|
|
||
| def create_label(self, label_data: "LabelUpdateParam") -> typing.Tuple[typing.Optional["LabelKeyData"], dict]: | ||
| return self.api_post("", label_data) | ||
| def create_label( | ||
| self, label_data: "LabelUpdateParam" | ||
| ) -> typing.Tuple[typing.Optional["LabelKeyData"], requests.models.Response]: | ||
| url = self.api_url | ||
| response = requests.post(url, json=label_data, timeout=TIMEOUT, headers=self._request_headers) | ||
| self._check_response(response) | ||
| return response.json(), response | ||
|
|
||
| def get_keys(self) -> typing.Tuple[typing.Optional["LabelsKeysData"], requests.models.Response]: | ||
| url = urljoin(self.api_url, "keys") | ||
|
|
||
| response = requests.get(url, timeout=TIMEOUT, headers=self._request_headers) | ||
| self._check_response(response) | ||
| return response.json(), response | ||
|
|
||
| def get_keys(self) -> typing.Tuple[typing.Optional["LabelsKeysData"], dict]: | ||
| return self.api_get("keys") | ||
| def get_values(self, key_id: str) -> typing.Tuple[typing.Optional["LabelKeyData"], requests.models.Response]: | ||
| url = urljoin(self.api_url, f"id/{key_id}") | ||
|
|
||
| def get_values(self, key_id: str) -> typing.Tuple[typing.Optional["LabelKeyData"], dict]: | ||
| return self.api_get(f"id/{key_id}") | ||
| response = requests.get(url, timeout=TIMEOUT, headers=self._request_headers) | ||
| self._check_response(response) | ||
| return response.json(), response | ||
|
|
||
| def get_value(self, key_id: str, value_id: str) -> typing.Tuple[typing.Optional["LabelKeyData"], dict]: | ||
| return self.api_get(f"id/{key_id}/values/{value_id}") | ||
| def get_value( | ||
| self, key_id: str, value_id: str | ||
| ) -> typing.Tuple[typing.Optional["LabelKeyData"], requests.models.Response]: | ||
| url = urljoin(self.api_url, f"id/{key_id}/values/{value_id}") | ||
|
|
||
| response = requests.get(url, timeout=TIMEOUT, headers=self._request_headers) | ||
| self._check_response(response) | ||
| return response.json(), response | ||
|
|
||
| def add_value( | ||
| self, key_id: str, label_data: "LabelUpdateParam" | ||
| ) -> typing.Tuple[typing.Optional["LabelKeyData"], dict]: | ||
| return self.api_post(f"id/{key_id}/values", label_data) | ||
| ) -> typing.Tuple[typing.Optional["LabelKeyData"], requests.models.Response]: | ||
| url = urljoin(self.api_url, f"id/{key_id}/values") | ||
|
|
||
| response = requests.post(url, json=label_data, timeout=TIMEOUT, headers=self._request_headers) | ||
| self._check_response(response) | ||
| return response.json(), response | ||
|
|
||
| def rename_key( | ||
| self, key_id: str, label_data: "LabelUpdateParam" | ||
| ) -> typing.Tuple[typing.Optional["LabelKeyData"], dict]: | ||
| return self.api_put(f"id/{key_id}", label_data) | ||
| ) -> typing.Tuple[typing.Optional["LabelKeyData"], requests.models.Response]: | ||
| url = urljoin(self.api_url, f"id/{key_id}") | ||
|
|
||
| response = requests.put(url, json=label_data, timeout=TIMEOUT, headers=self._request_headers) | ||
| self._check_response(response) | ||
| return response.json(), response | ||
|
|
||
| def rename_value( | ||
| self, key_id: str, value_id: str, label_data: "LabelUpdateParam" | ||
| ) -> typing.Tuple[typing.Optional["LabelKeyData"], dict]: | ||
| return self.api_put(f"id/{key_id}/values/{value_id}", label_data) | ||
| ) -> typing.Tuple[typing.Optional["LabelKeyData"], requests.models.Response]: | ||
| url = urljoin(self.api_url, f"id/{key_id}/values/{value_id}") | ||
|
|
||
| response = requests.put(url, json=label_data, timeout=TIMEOUT, headers=self._request_headers) | ||
| self._check_response(response) | ||
| return response.json(), response | ||
|
|
||
| def _check_response(self, response: requests.models.Response): | ||
| """ | ||
| Wraps an exceptional response to LabelsRepoAPIException | ||
| """ | ||
| message = None | ||
|
|
||
| if 400 <= response.status_code < 500: | ||
| error_data = response.json() | ||
| message = error_data.get("message", None) | ||
| elif 500 <= response.status_code < 600: | ||
| message = response.reason | ||
|
|
||
| if message: | ||
| raise LabelsRepoAPIException( | ||
| status=response.status_code, | ||
| url=response.request.url, | ||
| msg=message, | ||
| method=response.request.method, | ||
| ) | ||
|
|
||
| @property | ||
| def _request_headers(self): | ||
| return {"User-Agent": settings.GRAFANA_COM_USER_AGENT, "Authorization": f"Bearer {self.api_token}"} |