Allow setting the time of a point manually.

grundleborg · grundleborg · commit c25ec08926e3 · 2016-03-08T12:27:15.000Z
Point can be specified as either a number of nanoseconds, a python datetime object (with or without timezone) or a string in ISO datetime format. If a time is not specified, the Helper sets the time at the time of assembling the point fields so that multiple unique points with the same tags can be committed simultaneously without them failing to add due to all being assigned the same automatic time by the InfluxDB server. This fix is based upon the discussion in influxdata#130 but also includes the outstanding items for it to be merged. I'm happy to receive suggestions for further ways to add test coverage to this change. This also fixes influxdata#264 and fixes influxdata#259.
diff --git a/influxdb/helper.py b/influxdb/helper.py
@@ -3,6 +3,7 @@
 Helper class for InfluxDB
 """
 from collections import namedtuple, defaultdict
+from datetime import datetime
 from warnings import warn
 
 import six
@@ -16,6 +17,16 @@ class SeriesHelper(object):
     Each subclass can write to its own database.
     The time series names can also be based on one or more defined fields.
 
+    A field "time" can be used to write data points at a specific time,
+    rather than the default current time. The time field can take any of
+    the following forms:
+     * An integer unix timestamp in nanoseconds, assumed to be in UTC.
+     * A string in the ISO time format, including a timezone.
+     * A naive python datetime, which will be treated as UTC.
+     * A localized python datetime, which will use the chosen timezone.
+    If no time field is provided, the current UTC system time in microseconds
+    at the time of assembling the point data will be used.
+
     Annotated example::
 
         class MySeriesHelper(SeriesHelper):
@@ -142,8 +153,23 @@ def _json_body_(cls):
                     "tags": {},
                 }
 
+                ts = getattr(point, 'time', None)
+                if not ts:
+                    # No time provided. Use current UTC time.
+                    ts = datetime.utcnow().isoformat() + "+00:00"
+                elif isinstance(ts, datetime):
+                    if ts.tzinfo is None or ts.tzinfo.utcoffset(ts) is None:
+                        # Assuming naive datetime provided. Format with UTC tz.
+                        ts = ts.isoformat() + "+00:00"
+                    else:
+                        # Assuming localized datetime provided.
+                        ts = ts.isoformat()
+                # Neither of the above match. Assuming correct string or int.
+                json_point['time'] = ts
+
                 for field in cls._fields:
-                    json_point['fields'][field] = getattr(point, field)
+                    if field != 'time':
+                        json_point['fields'][field] = getattr(point, field)
 
                 for tag in cls._tags:
                     json_point['tags'][tag] = getattr(point, tag)
diff --git a/influxdb/tests/helper_test.py b/influxdb/tests/helper_test.py
@@ -1,5 +1,7 @@
 # -*- coding: utf-8 -*-
 
+import datetime
+import pytz
 import sys
 if sys.version_info < (2, 7):
     import unittest2 as unittest
@@ -38,6 +40,18 @@ class Meta:
 
         TestSeriesHelper.MySeriesHelper = MySeriesHelper
 
+        class MySeriesTimeHelper(SeriesHelper):
+
+            class Meta:
+                client = TestSeriesHelper.client
+                series_name = 'events.stats.{server_name}'
+                fields = ['time', 'some_stat']
+                tags = ['server_name', 'other_tag']
+                bulk_size = 5
+                autocommit = True
+
+        TestSeriesHelper.MySeriesTimeHelper = MySeriesTimeHelper
+
     def test_auto_commit(self):
         """
         Tests that write_points is called after the right number of events
@@ -66,14 +80,20 @@ def testSingleSeriesName(self):
         """
         Tests JSON conversion when there is only one series name.
         """
-        TestSeriesHelper.MySeriesHelper(
-            server_name='us.east-1', other_tag='ello', some_stat=159)
-        TestSeriesHelper.MySeriesHelper(
-            server_name='us.east-1', other_tag='ello', some_stat=158)
-        TestSeriesHelper.MySeriesHelper(
-            server_name='us.east-1', other_tag='ello', some_stat=157)
-        TestSeriesHelper.MySeriesHelper(
-            server_name='us.east-1', other_tag='ello', some_stat=156)
+        dt = datetime.datetime(2016, 1, 2, 3, 4, 5, 678912)
+        ts1 = dt
+        ts2 = "2016-10-11T01:02:03.123456789-04:00"
+        ts3 = 1234567890123456789
+        ts4 = pytz.timezone("Europe/Berlin").localize(dt)
+
+        TestSeriesHelper.MySeriesTimeHelper(
+            time=ts1, server_name='us.east-1', other_tag='ello', some_stat=159)
+        TestSeriesHelper.MySeriesTimeHelper(
+            time=ts2, server_name='us.east-1', other_tag='ello', some_stat=158)
+        TestSeriesHelper.MySeriesTimeHelper(
+            time=ts3, server_name='us.east-1', other_tag='ello', some_stat=157)
+        TestSeriesHelper.MySeriesTimeHelper(
+            time=ts4, server_name='us.east-1', other_tag='ello', some_stat=156)
         expectation = [
             {
                 "measurement": "events.stats.us.east-1",
@@ -84,6 +104,7 @@ def testSingleSeriesName(self):
                 "fields": {
                     "some_stat": 159
                 },
+                "time": "2016-01-02T03:04:05.678912+00:00",
             },
             {
                 "measurement": "events.stats.us.east-1",
@@ -94,6 +115,7 @@ def testSingleSeriesName(self):
                 "fields": {
                     "some_stat": 158
                 },
+                "time": "2016-10-11T01:02:03.123456789-04:00",
             },
             {
                 "measurement": "events.stats.us.east-1",
@@ -104,6 +126,7 @@ def testSingleSeriesName(self):
                 "fields": {
                     "some_stat": 157
                 },
+                "time": 1234567890123456789,
             },
             {
                 "measurement": "events.stats.us.east-1",
@@ -114,23 +137,24 @@ def testSingleSeriesName(self):
                 "fields": {
                     "some_stat": 156
                 },
+                "time": "2016-01-02T03:04:05.678912+01:00",
             }
         ]
 
-        rcvd = TestSeriesHelper.MySeriesHelper._json_body_()
+        rcvd = TestSeriesHelper.MySeriesTimeHelper._json_body_()
         self.assertTrue(all([el in expectation for el in rcvd]) and
                         all([el in rcvd for el in expectation]),
                         'Invalid JSON body of time series returned from '
                         '_json_body_ for one series name: {0}.'.format(rcvd))
-        TestSeriesHelper.MySeriesHelper._reset_()
+        TestSeriesHelper.MySeriesTimeHelper._reset_()
         self.assertEqual(
-            TestSeriesHelper.MySeriesHelper._json_body_(),
+            TestSeriesHelper.MySeriesTimeHelper._json_body_(),
             [],
             'Resetting helper did not empty datapoints.')
 
     def testSeveralSeriesNames(self):
         '''
-        Tests JSON conversion when there is only one series name.
+        Tests JSON conversion when there are multiple series names.
         '''
         TestSeriesHelper.MySeriesHelper(
             server_name='us.east-1', some_stat=159, other_tag='ello')
@@ -184,6 +208,10 @@ def testSeveralSeriesNames(self):
         ]
 
         rcvd = TestSeriesHelper.MySeriesHelper._json_body_()
+        for r in rcvd:
+            self.assertTrue(r.get('time'),
+                            "No time field in received JSON body.")
+            del(r["time"])
         self.assertTrue(all([el in expectation for el in rcvd]) and
                         all([el in rcvd for el in expectation]),
                         'Invalid JSON body of time series returned from '
diff --git a/test-requirements.txt b/test-requirements.txt
@@ -1,4 +1,5 @@
 nose
 nose-cov
 mock
-requests-mock
+requests-mock
+pytz
