1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
|
"""Exceptions utils for InfluxDB."""
import logging
from urllib3 import HTTPResponse
logger = logging.getLogger('influxdb_client.client.exceptions')
class InfluxDBError(Exception):
"""Raised when a server error occurs."""
def __init__(self, response: HTTPResponse = None, message: str = None):
"""Initialize the InfluxDBError handler."""
if response is not None:
self.response = response
self.message = self._get_message(response)
self.retry_after = response.getheader('Retry-After')
else:
self.response = None
self.message = message or 'no response'
self.retry_after = None
super().__init__(self.message)
def _get_message(self, response):
# Body
if response.data:
import json
try:
return json.loads(response.data)["message"]
except Exception as e:
logging.debug(f"Cannot parse error response to JSON: {response.data}, {e}")
return response.data
# Header
for header_key in ["X-Platform-Error-Code", "X-Influx-Error", "X-InfluxDb-Error"]:
header_value = response.getheader(header_key)
if header_value is not None:
return header_value
# Http Status
return response.reason
|
