Line protocol

.travis.yml

@@ -12,7 +12,7 @@ env:
 install:
     - sudo pip install tox
     - sudo pip install coveralls
-    - wget http://get.influxdb.org/influxdb_0.9.0_amd64.deb && sudo dpkg -i influxdb_0.9.0_amd64.deb
+    - wget https://s3.amazonaws.com/influxdb/influxdb_0.9.1_amd64.deb && sudo dpkg -i influxdb_0.9.1_amd64.deb
 script:
     - travis_wait 30 tox -e $TOX_ENV
 after_success:

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
 from .exceptions import InfluxDBClientError
 from .exceptions import InfluxDBServerError
@@ -96,7 +97,8 @@ def __init__(self,
 
         self._headers = {
             'Content-type': 'application/json',
-            'Accept': 'text/plain'}
+            'Accept': 'text/plain'
+        }
 
     @staticmethod
     def from_DSN(dsn, **kwargs):
@@ -182,7 +184,7 @@ def switch_user(self, username, password):
         self._password = password
 
     def request(self, url, method='GET', params=None, data=None,
-                expected_response_code=200):
+                expected_response_code=200, headers=None):
         """Make a HTTP request to the InfluxDB API.
 
         :param url: the path of the HTTP request, e.g. write, query, etc.
@@ -203,17 +205,13 @@ def request(self, url, method='GET', params=None, data=None,
         """
         url = "{0}/{1}".format(self._baseurl, url)
 
+        if headers is None:
+            headers = self._headers
+
         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, list)):
             data = json.dumps(data)
 
         # Try to send the request a maximum of three times. (see #103)
@@ -223,9 +221,10 @@ 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,
+                    headers=headers,
                     verify=self._verify_ssl,
                     timeout=self._timeout
                 )
@@ -254,18 +253,24 @@ def write(self, data, params=None, expected_response_code=204):
         :returns: True, if the write operation is successful
         :rtype: bool
         """
+
+        headers = self._headers
+        headers['Content-type'] = 'application/octet-stream'
+
         self.request(
             url="write",
             method='POST',
             params=params,
-            data=data,
-            expected_response_code=expected_response_code
+            data=make_lines(data).encode('utf-8'),
+            expected_response_code=expected_response_code,
+            headers=headers
         )
         return True
 
     def query(self,
               query,
               params={},
+              epoch=None,
               expected_response_code=200,
               database=None,
               raise_errors=True):
@@ -294,6 +299,9 @@ def query(self,
         params['q'] = query
         params['db'] = database or self._database
 
+        if epoch is not None:
+            params['epoch'] = epoch
+
         response = self.request(
             url="query",
             method='GET',
@@ -391,22 +399,25 @@ def _write_points(self,
             'points': points
         }
 
-        if time_precision:
-            data['precision'] = time_precision
+        if tags is not None:
+            data['tags'] = tags
 
-        if retention_policy:
-            data['retentionPolicy'] = retention_policy
+        params = {
+            'db': database or self._database
+        }
 
-        if tags:
-            data['tags'] = tags
+        if time_precision is not None:
+            params['precision'] = time_precision
 
-        data['database'] = database or self._database
+        if retention_policy is not None:
+            params['rp'] = retention_policy
 
         if self.use_udp:
             self.send_packet(data)
         else:
             self.write(
                 data=data,
+                params=params,
                 expected_response_code=204
             )
 
@@ -578,15 +589,20 @@ def get_list_users(self):
         """
         return list(self.query("SHOW USERS").get_points())
 
-    def create_user(self, username, password):
+    def create_user(self, username, password, admin=False):
         """Create a new user in InfluxDB
 
         :param username: the new username to create
         :type username: str
         :param password: the password for the new user
         :type password: str
+        :param admin: whether the user should have cluster administration
+            privileges or not
+        :type admin: boolean
         """
         text = "CREATE USER {} WITH PASSWORD '{}'".format(username, password)
+        if admin:
+            text += ' WITH ALL PRIVILEGES'
         self.query(text)
 
     def drop_user(self, username):
@@ -609,29 +625,27 @@ def set_user_password(self, username, password):
         text = "SET PASSWORD FOR {} = '{}'".format(username, password)
         self.query(text)
 
-    def delete_series(self, id, database=None):
-        """Delete series from a database.
+    def delete_series(self, database=None, measurement=None, tags=None):
+        """Delete series from a database. Series can be filtered by
+        measurement and tags.
 
-        :param id: the id of the series to be deleted
-        :type id: int
+        :param measurement: Delete all series from a measurement
+        :type id: string
+        :param tags: Delete all series that match given tags
+        :type id: dict
         :param database: the database from which the series should be
             deleted, defaults to client's current database
         :type database: str
         """
         database = database or self._database
-        self.query('DROP SERIES %s' % id, database=database)
-
-    def grant_admin_privileges(self, username):
-        """Grant cluster administration privileges to an user.
-
-        :param username: the username to grant privileges to
-        :type username: str
+        query_str = 'DROP SERIES'
+        if measurement:
+            query_str += ' FROM "{}"'.format(measurement)
 
-        .. note:: Only a cluster administrator can create/ drop databases
-            and manage users.
-        """
-        text = "GRANT ALL PRIVILEGES TO {}".format(username)
-        self.query(text)
+        if tags:
+            query_str += ' WHERE ' + ' and '.join(["{}='{}'".format(k, v)
+                                                   for k, v in tags.items()])
+        self.query(query_str, database=database)
 
     def revoke_admin_privileges(self, username):
         """Revoke cluster administration privileges from an user.
@@ -683,9 +697,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):

influxdb/line_protocol.py

@@ -0,0 +1,103 @@
+# -*- coding: utf-8 -*-
+from __future__ import unicode_literals
+
+from calendar import timegm
+from copy import copy
+from datetime import datetime
+
+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 (
+            timegm(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')
+    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 'time' in point:
+            lines += " " + _force_text(str(int(
+                _convert_timestamp(point['time'])
+            )))
+
+        lines += "\n"
+    return lines

requirements.txt

@@ -1,2 +1,4 @@
+python-dateutil>=2.0.0
+pytz
 requests>=1.0.3
-six==1.9.0
+six==1.9.0