Fixes/msgpack default

influxdb/client.py

@@ -22,7 +22,6 @@
 from influxdb.line_protocol import make_lines, quote_ident, quote_literal
 from influxdb.resultset import ResultSet
 from .exceptions import InfluxDBClientError
-from .exceptions import InfluxDBServerError
 
 
 class InfluxDBClient(object):
@@ -89,6 +88,7 @@ def __init__(self,
                  pool_size=10,
                  path='',
                  cert=None,
+                 use_msgpack=False
                  ):
         """Construct a new InfluxDBClient object."""
         self.__host = host
@@ -110,7 +110,9 @@ def __init__(self,
         )
 
         if use_udp:
-            self.udp_socket = socket.socket(socket.AF_INET, socket.SOCK_DGRAM)
+            self._udp_socket = socket.socket(socket.AF_INET, socket.SOCK_DGRAM)
+        else:
+            self._udp_socket = None
 
         if not path:
             self.__path = ''
@@ -145,10 +147,16 @@ def __init__(self,
             self._port,
             self._path)
 
-        self._headers = {
-            'Content-Type': 'application/json',
-            'Accept': 'application/x-msgpack'
-        }
+        if use_msgpack:
+            self._headers = {
+                'Content-Type': 'application/json',
+                'Accept': 'application/x-msgpack'
+            }
+        else:
+            self._headers = {
+                'Content-Type': 'application/json',
+                'Accept': 'text/plain'
+            }
 
     @property
     def _baseurl(self):
@@ -243,14 +251,14 @@ def request(self, url, method='GET', params=None, data=None,
         :param method: the HTTP method for the request, defaults to GET
         :type method: str
         :param params: additional parameters for the request, defaults to None
-        :type params: dict
+        :type params: dict, optional
         :param data: the data of the request, defaults to None
-        :type data: str
+        :type data: str, optional
         :param expected_response_code: the expected response code of
             the request, defaults to 200
         :type expected_response_code: int
         :param headers: headers to add to the request
-        :type headers: dict
+        :type headers: dict, optional
         :returns: the response from the request
         :rtype: :class:`requests.Response`
         :raises InfluxDBServerError: if the response code is any server error
@@ -285,6 +293,7 @@ def request(self, url, method='GET', params=None, data=None,
                     verify=self._verify_ssl,
                     timeout=self._timeout
                 )
+                response._msgpack = None
                 break
             except (requests.exceptions.ConnectionError,
                     requests.exceptions.HTTPError,
@@ -297,30 +306,39 @@ def request(self, url, method='GET', params=None, data=None,
                 if not retry:
                     raise
 
-        type_header = response.headers and response.headers.get("Content-Type")
-        if type_header == "application/x-msgpack" and response.content:
-            response._msgpack = msgpack.unpackb(
-                packed=response.content,
-                ext_hook=_msgpack_parse_hook,
-                raw=False)
-        else:
-            response._msgpack = None
+        if self._is_msg_pack_response(response):
+            if response.content:
+                response._msgpack = msgpack.unpackb(
+                    packed=response.content,
+                    ext_hook=_msgpack_parse_hook,
+                    raw=False)
 
-        def reformat_error(response):
-            if response._msgpack:
-                return json.dumps(response._msgpack, separators=(',', ':'))
-            else:
-                return response.content
-
-        # if there's not an error, there must have been a successful response
-        if 500 <= response.status_code < 600:
-            raise InfluxDBServerError(reformat_error(response))
-        elif response.status_code == expected_response_code:
+        if response.status_code == expected_response_code:
             return response
         else:
-            err_msg = reformat_error(response)
+            err_msg = self._reformat_msgpack_error(response)
             raise InfluxDBClientError(err_msg, response.status_code)
 
+    @staticmethod
+    def _is_msg_pack_response(response):
+        if response is None:
+            return False
+
+        if response.headers is None:
+            return False
+
+        if "Content-Type" not in response.headers:
+            return False
+
+        content_type = response.headers["Content-Type"]
+        return content_type == "application/x-msgpack"
+
+    def _reformat_msgpack_error(self, _response):
+        if _response._msgpack is not None:
+            return json.dumps(_response._msgpack, separators=(',', ':'))
+        else:
+            return _response.content
+
     def write(self, data, params=None, expected_response_code=204,
               protocol='json'):
         """Write data to InfluxDB.
@@ -697,7 +715,7 @@ def create_retention_policy(self, name, duration, replication,
             The minimum retention period is 1 hour.
         :type duration: str
         :param replication: the replication of the retention policy
-        :type replication: str
+        :type replication: int
         :param database: the database for which the retention policy is
             created. Defaults to current client's database
         :type database: str
@@ -717,7 +735,7 @@ def create_retention_policy(self, name, duration, replication,
             "CREATE RETENTION POLICY {0} ON {1} " \
             "DURATION {2} REPLICATION {3} SHARD DURATION {4}".format(
                 quote_ident(name), quote_ident(database or self._database),
-                duration, replication, shard_duration)
+                duration, int(replication), shard_duration)
 
         if default is True:
             query_string += " DEFAULT"
@@ -1071,7 +1089,7 @@ def drop_continuous_query(self, name, database=None):
         self.query(query_string)
 
     def send_packet(self, packet, protocol='json', time_precision=None):
-        """Send an UDP packet.
+        """Send an UDP packet.  Only valid when use_udp is True.
 
         :param packet: the packet to be sent
         :type packet: (if protocol is 'json') dict
@@ -1081,11 +1099,18 @@ def send_packet(self, packet, protocol='json', time_precision=None):
         :param time_precision: Either 's', 'm', 'ms' or 'u', defaults to None
         :type time_precision: str
         """
+        if not self._use_udp:
+            raise RuntimeError("Unable to send packet : use_udp set to False")
+
         if protocol == 'json':
             data = make_lines(packet, time_precision).encode('utf-8')
         elif protocol == 'line':
             data = ('\n'.join(packet) + '\n').encode('utf-8')
-        self.udp_socket.sendto(data, (self._host, self._udp_port))
+        else:
+            raise InfluxDBClientError("Invalid protocol name : "
+                                      "expected json or line")
+
+        self._udp_socket.sendto(data, (self._host, self._udp_port))
 
     def close(self):
         """Close http session."""

influxdb/resultset.py

@@ -129,9 +129,9 @@ def __iter__(self):
             yield list(self.__getitem__(key))
 
     @staticmethod
-    def _tag_matches(tags, filter):
+    def _tag_matches(tags, _filter):
         """Check if all key/values in filter match in tags."""
-        for tag_name, tag_value in filter.items():
+        for tag_name, tag_value in _filter.items():
             # using _sentinel as I'm not sure that "None"
             # could be used, because it could be a valid
             # series_tags value : when a series has no such tag