initial implementation of line protocol

trehn · trehn · commit 078350e35f43 · 2015-06-21T12:08:32.000+02:00
as introduced in influxdata/influxdb#2696
diff --git a/influxdb/client.py b/influxdb/client.py
@@ -11,6 +11,7 @@
 import requests.exceptions
 from sys import version_info
 
+from influxdb.line_protocol import make_lines
 from influxdb.resultset import ResultSet
 
 try:
@@ -221,14 +222,7 @@ def request(self, url, method='GET', params=None, data=None,
         if params is None:
             params = {}
 
-        auth = {
-            'u': self._username,
-            'p': self._password
-        }
-
-        params.update(auth)
-
-        if data is not None and not isinstance(data, str):
+        if isinstance(data, dict) or isinstance(data, list):
             data = json.dumps(data)
 
         # Try to send the request a maximum of three times. (see #103)
@@ -238,6 +232,7 @@ def request(self, url, method='GET', params=None, data=None,
                 response = self._session.request(
                     method=method,
                     url=url,
+                    auth=(self._username, self._password),
                     params=params,
                     data=data,
                     headers=self._headers,
@@ -270,10 +265,10 @@ def write(self, data, params=None, expected_response_code=204):
         :rtype: bool
         """
         self.request(
-            url="write",
+            url="write_points",
             method='POST',
             params=params,
-            data=data,
+            data=make_lines(data).encode('utf-8'),
             expected_response_code=expected_response_code
         )
         return True
@@ -396,13 +391,12 @@ def _write_points(self,
         if tags:
             data['tags'] = tags
 
-        data['database'] = database or self._database
-
         if self.use_udp:
             self.send_packet(data)
         else:
             self.write(
                 data=data,
+                params={'db': database or self._database},
                 expected_response_code=204
             )
 
@@ -679,9 +673,8 @@ def send_packet(self, packet):
         :param packet: the packet to be sent
         :type packet: dict
         """
-        data = json.dumps(packet)
-        byte = data.encode('utf-8')
-        self.udp_socket.sendto(byte, (self._host, self.udp_port))
+        data = make_lines(packet).encode('utf-8')
+        self.udp_socket.sendto(data, (self._host, self.udp_port))
 
 
 class InfluxDBClusterClient(object):
diff --git a/influxdb/line_protocol.py b/influxdb/line_protocol.py
@@ -0,0 +1,103 @@
+# -*- coding: utf-8 -*-
+from __future__ import unicode_literals
+
+from copy import copy
+from datetime import datetime
+from time import mktime
+
+from dateutil.parser import parse
+from pytz import utc
+from six import binary_type, text_type
+
+
+def _convert_timestamp(timestamp):
+    if isinstance(timestamp, int):
+        return timestamp
+    if isinstance(_force_text(timestamp), text_type):
+        timestamp = parse(timestamp)
+    if isinstance(timestamp, datetime):
+        if timestamp.tzinfo:
+            timestamp = timestamp.astimezone(utc)
+            timestamp.replace(tzinfo=None)
+        return (
+            mktime(timestamp.timetuple()) * 1e9 +
+            timestamp.microsecond * 1e3
+        )
+    raise ValueError(timestamp)
+
+
+def _escape_tag(tag):
+    return tag.replace(
+        "\\", "\\\\"
+    ).replace(
+        " ", "\\ "
+    ).replace(
+        ",", "\\,"
+    ).replace(
+        "=", "\\="
+    )
+
+
+def _escape_value(value):
+    value = _force_text(value)
+    if isinstance(value, text_type):
+        return "\"{}\"".format(value.replace(
+            "\"", "\\\""
+        ))
+    else:
+        return str(value)
+
+
+def _force_text(data):
+    """
+    Try to return a text aka unicode object from the given data.
+    """
+    if isinstance(data, binary_type):
+        return data.decode('utf-8', 'replace')
+    else:
+        return data
+
+
+def make_lines(data):
+    """
+    Extracts the points from the given dict and returns a Unicode string
+    matching the line protocol introduced in InfluxDB 0.9.0.
+    """
+    lines = ""
+    static_tags = data.get('tags', None)
+    for point in data['points']:
+        # add measurement name
+        lines += _escape_tag(_force_text(
+            point.get('measurement', data.get('measurement'))
+        )) + ","
+
+        # add tags
+        if static_tags is None:
+            tags = point.get('tags', {})
+        else:
+            tags = copy(static_tags)
+            tags.update(point.get('tags', {}))
+        # tags should be sorted client-side to take load off server
+        for tag_key in sorted(tags.keys()):
+            lines += "{key}={value},".format(
+                key=_escape_tag(tag_key),
+                value=_escape_tag(tags[tag_key]),
+            )
+        lines = lines[:-1] + " "  # strip the trailing comma
+
+        # add fields
+        for field_key in sorted(point['fields'].keys()):
+            lines += "{key}={value},".format(
+                key=_escape_tag(field_key),
+                value=_escape_value(point['fields'][field_key]),
+            )
+        lines = lines[:-1]  # strip the trailing comma
+
+        # add timestamp
+        if 'timestamp' in point:
+            lines += " " + _force_text(str(int(
+                _convert_timestamp(point['timestamp'])
+            )))
+
+        lines += "\n"
+    return lines
diff --git a/requirements.txt b/requirements.txt
@@ -1,2 +1,4 @@
+python-dateutil>=2.0.0
+pytz
 requests>=1.0.3
-six==1.9.0
+six==1.9.0
