|
| 1 | +import json |
| 2 | +import logging |
| 3 | + |
| 4 | +from .model import ( |
| 5 | + APIModel, |
| 6 | + APIEndpoints, |
| 7 | + RequestsMethods, |
| 8 | +) |
| 9 | +from .api import Api |
| 10 | + |
| 11 | + |
| 12 | +class Snapshot: |
| 13 | + """The class includes all necessary methods to access the Grafana snapshot API endpoints |
| 14 | +
|
| 15 | + Args: |
| 16 | + grafana_api_model (APIModel): Inject a Grafana API model object that includes all necessary values and information |
| 17 | +
|
| 18 | + Attributes: |
| 19 | + grafana_api_model (APIModel): This is where we store the grafana_api_model |
| 20 | + """ |
| 21 | + |
| 22 | + def __init__(self, grafana_api_model: APIModel): |
| 23 | + self.grafana_api_model = grafana_api_model |
| 24 | + |
| 25 | + def create_new_snapshot( |
| 26 | + self, |
| 27 | + dashboard_json: dict, |
| 28 | + name: str = None, |
| 29 | + expires: int = None, |
| 30 | + external: bool = False, |
| 31 | + key: str = None, |
| 32 | + delete_key: str = None, |
| 33 | + ) -> dict: |
| 34 | + """The method includes a functionality to create the specified dashboard snapshot |
| 35 | +
|
| 36 | + Args: |
| 37 | + dashboard_json (dict): Specify the dashboard_json of the dashboard |
| 38 | + name (str): Specify the optional name of the dashboard snapshot |
| 39 | + expires (int): Specify the optional expiry time as seconds of the dashboard snapshot. 3600 is 1 hour, 86400 is 1 day (default never expired) |
| 40 | + external (bool): Specify the optional external server rather than locally (default False) |
| 41 | + key (str): Specify the optional unique key. Required if external is true. |
| 42 | + delete_key (str): Specify the optional unique key used to delete the snapshot. It is different from the key so that only the creator can delete the snapshot. Required if external is true. |
| 43 | +
|
| 44 | + Raises: |
| 45 | + ValueError: Missed specifying a necessary value |
| 46 | + Exception: Unspecified error by executing the API call |
| 47 | +
|
| 48 | + Returns: |
| 49 | + api_call (dict): Returns the snapshot information of the dashboard |
| 50 | + """ |
| 51 | + |
| 52 | + if dashboard_json != dict(): |
| 53 | + if external: |
| 54 | + if (key is None or len(key) == 0) and ( |
| 55 | + delete_key is None or len(delete_key) == 0 |
| 56 | + ): |
| 57 | + logging.error( |
| 58 | + "It's necessary that you define the key and the delete_key, if you use the external snapshot " |
| 59 | + "opportunity. " |
| 60 | + ) |
| 61 | + raise ValueError |
| 62 | + |
| 63 | + snapshot_json: dict = { |
| 64 | + "dashboard": dashboard_json, |
| 65 | + "name": name, |
| 66 | + "expires": expires, |
| 67 | + "external": external, |
| 68 | + "key": key, |
| 69 | + "deleteKey": delete_key, |
| 70 | + } |
| 71 | + |
| 72 | + api_call: dict = ( |
| 73 | + Api(self.grafana_api_model) |
| 74 | + .call_the_api( |
| 75 | + APIEndpoints.SNAPSHOTS.value, |
| 76 | + RequestsMethods.POST, |
| 77 | + json.dumps(snapshot_json), |
| 78 | + ) |
| 79 | + .json() |
| 80 | + ) |
| 81 | + |
| 82 | + if api_call == dict() or api_call.get("id") is None: |
| 83 | + logging.error(f"Check the error: {api_call}.") |
| 84 | + raise Exception |
| 85 | + else: |
| 86 | + return api_call |
| 87 | + else: |
| 88 | + logging.error("There is no dashboard_json defined.") |
| 89 | + raise ValueError |
| 90 | + |
| 91 | + def get_snapshots(self) -> list: |
| 92 | + """The method includes a functionality to list all dashboard snapshots |
| 93 | +
|
| 94 | + Raises: |
| 95 | + Exception: Unspecified error by executing the API call |
| 96 | +
|
| 97 | + Returns: |
| 98 | + api_call (list): Returns all dashboard snapshots |
| 99 | + """ |
| 100 | + |
| 101 | + api_call: list = ( |
| 102 | + Api(self.grafana_api_model) |
| 103 | + .call_the_api( |
| 104 | + APIEndpoints.DASHBOARD_SNAPSHOTS.value, |
| 105 | + ) |
| 106 | + .json() |
| 107 | + ) |
| 108 | + |
| 109 | + if api_call == list() or api_call[0].get("id") is None: |
| 110 | + logging.error(f"Check the error: {api_call}.") |
| 111 | + raise Exception |
| 112 | + else: |
| 113 | + return api_call |
| 114 | + |
| 115 | + def get_snapshot_by_key(self, key: str) -> dict: |
| 116 | + """The method includes a functionality to get a specific dashboard snapshot by the key |
| 117 | +
|
| 118 | + Args: |
| 119 | + key (str): Specify the key of the dashboard snapshot |
| 120 | +
|
| 121 | + Raises: |
| 122 | + ValueError: Missed specifying a necessary value |
| 123 | + Exception: Unspecified error by executing the API call |
| 124 | +
|
| 125 | + Returns: |
| 126 | + api_call (dict): Returns a specific dashboard snapshot |
| 127 | + """ |
| 128 | + |
| 129 | + if len(key) != 0: |
| 130 | + api_call: dict = ( |
| 131 | + Api(self.grafana_api_model) |
| 132 | + .call_the_api( |
| 133 | + f"{APIEndpoints.SNAPSHOTS.value}/{key}", |
| 134 | + ) |
| 135 | + .json() |
| 136 | + ) |
| 137 | + |
| 138 | + if api_call == dict() or api_call.get("dashboard").get("id") is None: |
| 139 | + logging.error(f"Check the error: {api_call}.") |
| 140 | + raise Exception |
| 141 | + else: |
| 142 | + return api_call |
| 143 | + else: |
| 144 | + logging.error("There is no key defined.") |
| 145 | + raise ValueError |
| 146 | + |
| 147 | + def delete_snapshot_by_key(self, key: str): |
| 148 | + """The method includes a functionality to delete a specific dashboard snapshot by the key |
| 149 | +
|
| 150 | + Args: |
| 151 | + key (str): Specify the key of the dashboard snapshot |
| 152 | +
|
| 153 | + Raises: |
| 154 | + ValueError: Missed specifying a necessary value |
| 155 | + Exception: Unspecified error by executing the API call |
| 156 | +
|
| 157 | + Returns: |
| 158 | + None |
| 159 | + """ |
| 160 | + |
| 161 | + if len(key) != 0: |
| 162 | + api_call: dict = ( |
| 163 | + Api(self.grafana_api_model) |
| 164 | + .call_the_api( |
| 165 | + f"{APIEndpoints.SNAPSHOTS.value}/{key}", RequestsMethods.DELETE |
| 166 | + ) |
| 167 | + .json() |
| 168 | + ) |
| 169 | + |
| 170 | + if ( |
| 171 | + api_call.get("message") |
| 172 | + != "Snapshot deleted. It might take an hour before it's cleared from any CDN caches." |
| 173 | + ): |
| 174 | + logging.error(f"Check the error: {api_call}.") |
| 175 | + raise Exception |
| 176 | + else: |
| 177 | + logging.info("You successfully destroyed the dashboard snapshot.") |
| 178 | + else: |
| 179 | + logging.error("There is no key defined.") |
| 180 | + raise ValueError |
| 181 | + |
| 182 | + def delete_snapshot_by_delete_key(self, delete_key: str): |
| 183 | + """The method includes a functionality to delete a specific dashboard snapshot by the delete_key |
| 184 | +
|
| 185 | + Args: |
| 186 | + delete_key (str): Specify the delete_key of the dashboard snapshot |
| 187 | +
|
| 188 | + Raises: |
| 189 | + ValueError: Missed specifying a necessary value |
| 190 | + Exception: Unspecified error by executing the API call |
| 191 | +
|
| 192 | + Returns: |
| 193 | + None |
| 194 | + """ |
| 195 | + |
| 196 | + if len(delete_key) != 0: |
| 197 | + api_call: dict = ( |
| 198 | + Api(self.grafana_api_model) |
| 199 | + .call_the_api( |
| 200 | + f"{APIEndpoints.SNAPSHOTS.value}-delete/{delete_key}", |
| 201 | + ) |
| 202 | + .json() |
| 203 | + ) |
| 204 | + |
| 205 | + if ( |
| 206 | + api_call.get("message") |
| 207 | + != "Snapshot deleted. It might take an hour before it's cleared from any CDN caches." |
| 208 | + ): |
| 209 | + logging.error(f"Check the error: {api_call}.") |
| 210 | + raise Exception |
| 211 | + else: |
| 212 | + logging.info("You successfully destroyed the dashboard snapshot.") |
| 213 | + else: |
| 214 | + logging.error("There is no delete_key defined.") |
| 215 | + raise ValueError |
0 commit comments