From d54a1102f903a0b6c7a1560a5b4bfb4d494d88ba Mon Sep 17 00:00:00 2001 From: Sebastian Borza Date: Tue, 7 Apr 2020 17:30:49 -0500 Subject: [PATCH 01/37] feat(client): re-add support for 'show series' from legacy PR #357. Closes #353 (#806) * feat(client): re-add support for 'show series' from legacy PR #357. Closes #353 * chore(client): fix failing tests * chore(client): update linters to pass --- CHANGELOG.md | 7 +++ influxdb/client.py | 35 +++++++++++ influxdb/tests/client_test.py | 60 +++++++++++++++++++ .../server_tests/client_test_with_server.py | 58 ++++++++++++++++++ 4 files changed, 160 insertions(+) diff --git a/CHANGELOG.md b/CHANGELOG.md index 13445e97..bdc08569 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -6,9 +6,16 @@ and this project adheres to [Semantic Versioning](http://semver.org/spec/v2.0.0. ## [Unreleased] +### Added + +### Changed + +## [v5.2.4] - 2020-04-10 + ### Added - Add mypy testing framework (#756) - Add support for messagepack (#734 thx @lovasoa) +- Add support for 'show series' (#357 thx @gaker) ### Changed - Clean up stale CI config (#755) diff --git a/influxdb/client.py b/influxdb/client.py index 5e39f490..43427a11 100644 --- a/influxdb/client.py +++ b/influxdb/client.py @@ -7,6 +7,7 @@ from __future__ import unicode_literals import datetime +import itertools import json import random import socket @@ -637,6 +638,40 @@ def get_list_database(self): """ return list(self.query("SHOW DATABASES").get_points()) + def get_list_series(self, database=None, measurement=None, tags=None): + """ + Query SHOW SERIES returns the distinct series in your database. + + FROM and WHERE clauses are optional. + + :param measurement: Show all series from a measurement + :type id: string + :param tags: Show all series that match given tags + :type id: dict + :param database: the database from which the series should be + shows, defaults to client's current database + :type database: str + """ + database = database or self._database + query_str = 'SHOW SERIES' + + if measurement: + query_str += ' FROM "{0}"'.format(measurement) + + if tags: + query_str += ' WHERE ' + ' and '.join(["{0}='{1}'".format(k, v) + for k, v in tags.items()]) + + return list( + itertools.chain.from_iterable( + [ + x.values() + for x in (self.query(query_str, database=database) + .get_points()) + ] + ) + ) + def create_database(self, dbname): """Create a new database in InfluxDB. diff --git a/influxdb/tests/client_test.py b/influxdb/tests/client_test.py index 54116f7e..99a7f42b 100644 --- a/influxdb/tests/client_test.py +++ b/influxdb/tests/client_test.py @@ -689,6 +689,66 @@ def test_get_list_measurements(self): [{'name': 'cpu'}, {'name': 'disk'}] ) + def test_get_list_series(self): + """Test get a list of series from the database.""" + data = {'results': [ + {'series': [ + { + 'values': [ + ['cpu_load_short,host=server01,region=us-west'], + ['memory_usage,host=server02,region=us-east']], + 'columns': ['key'] + } + ]} + ]} + + with _mocked_session(self.cli, 'get', 200, json.dumps(data)): + self.assertListEqual( + self.cli.get_list_series(), + ['cpu_load_short,host=server01,region=us-west', + 'memory_usage,host=server02,region=us-east']) + + def test_get_list_series_with_measurement(self): + """Test get a list of series from the database by filter.""" + data = {'results': [ + {'series': [ + { + 'values': [ + ['cpu_load_short,host=server01,region=us-west']], + 'columns': ['key'] + } + ]} + ]} + + with _mocked_session(self.cli, 'get', 200, json.dumps(data)): + self.assertListEqual( + self.cli.get_list_series(measurement='cpu_load_short'), + ['cpu_load_short,host=server01,region=us-west']) + + def test_get_list_series_with_tags(self): + """Test get a list of series from the database by tags.""" + data = {'results': [ + {'series': [ + { + 'values': [ + ['cpu_load_short,host=server01,region=us-west']], + 'columns': ['key'] + } + ]} + ]} + + with _mocked_session(self.cli, 'get', 200, json.dumps(data)): + self.assertListEqual( + self.cli.get_list_series(tags={'region': 'us-west'}), + ['cpu_load_short,host=server01,region=us-west']) + + @raises(Exception) + def test_get_list_series_fails(self): + """Test get a list of series from the database but fail.""" + cli = InfluxDBClient('host', 8086, 'username', 'password') + with _mocked_session(cli, 'get', 401): + cli.get_list_series() + def test_create_retention_policy_default(self): """Test create default ret policy for TestInfluxDBClient object.""" example_response = '{"results":[{}]}' diff --git a/influxdb/tests/server_tests/client_test_with_server.py b/influxdb/tests/server_tests/client_test_with_server.py index fda3f720..94f28b66 100644 --- a/influxdb/tests/server_tests/client_test_with_server.py +++ b/influxdb/tests/server_tests/client_test_with_server.py @@ -817,6 +817,64 @@ def test_query_multiple_series(self): ] self.cli.write_points(pts) + def test_get_list_series(self): + """Test get a list of series from the database.""" + dummy_points = [ + { + "measurement": "cpu_load_short", + "tags": { + "host": "server01", + "region": "us-west" + }, + "time": "2009-11-10T23:00:00.123456Z", + "fields": { + "value": 0.64 + } + } + ] + + dummy_points_2 = [ + { + "measurement": "memory_usage", + "tags": { + "host": "server02", + "region": "us-east" + }, + "time": "2009-11-10T23:00:00.123456Z", + "fields": { + "value": 80 + } + } + ] + + self.cli.write_points(dummy_points) + self.cli.write_points(dummy_points_2) + + self.assertEquals( + self.cli.get_list_series(), + ['cpu_load_short,host=server01,region=us-west', + 'memory_usage,host=server02,region=us-east'] + ) + + self.assertEquals( + self.cli.get_list_series(measurement='memory_usage'), + ['memory_usage,host=server02,region=us-east'] + ) + + self.assertEquals( + self.cli.get_list_series(measurement='memory_usage'), + ['memory_usage,host=server02,region=us-east'] + ) + + self.assertEquals( + self.cli.get_list_series(tags={'host': 'server02'}), + ['memory_usage,host=server02,region=us-east']) + + self.assertEquals( + self.cli.get_list_series( + measurement='cpu_load_short', tags={'host': 'server02'}), + []) + @skip_server_tests class UdpTests(ManyTestCasesWithServerMixin, unittest.TestCase): From 794f0d819f312b344d14654abe55c36b4b973d7b Mon Sep 17 00:00:00 2001 From: Sebastian Borza Date: Tue, 7 Apr 2020 18:36:21 -0500 Subject: [PATCH 02/37] feat(client): allow custom requests session in InfluxDBClient (#807) * feat(client): enable client request to provide custom requests session * feat(client): allow custom requests session in InfluxDBClient --- CHANGELOG.md | 1 + influxdb/client.py | 10 +++++++++- 2 files changed, 10 insertions(+), 1 deletion(-) diff --git a/CHANGELOG.md b/CHANGELOG.md index bdc08569..59777b5f 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -16,6 +16,7 @@ and this project adheres to [Semantic Versioning](http://semver.org/spec/v2.0.0. - Add mypy testing framework (#756) - Add support for messagepack (#734 thx @lovasoa) - Add support for 'show series' (#357 thx @gaker) +- Add support for custom request session in InfluxDBClient (#360 thx @dschien) ### Changed - Clean up stale CI config (#755) diff --git a/influxdb/client.py b/influxdb/client.py index 43427a11..390d8e16 100644 --- a/influxdb/client.py +++ b/influxdb/client.py @@ -70,6 +70,9 @@ class InfluxDBClient(object): as a single file containing the private key and the certificate, or as a tuple of both files’ paths, defaults to None :type cert: str + :param session: allow for the new client request to use an existing + requests Session, defaults to None + :type session: requests.Session :raises ValueError: if cert is provided but ssl is disabled (set to False) """ @@ -90,6 +93,7 @@ def __init__(self, pool_size=10, path='', cert=None, + session=None, ): """Construct a new InfluxDBClient object.""" self.__host = host @@ -104,7 +108,11 @@ def __init__(self, self.__use_udp = use_udp self.__udp_port = udp_port - self._session = requests.Session() + + if not session: + session = requests.Session() + + self._session = session adapter = requests.adapters.HTTPAdapter( pool_connections=int(pool_size), pool_maxsize=int(pool_size) From 68b27ac1e0181c02e63d26985bdbb9a6fbcff1ff Mon Sep 17 00:00:00 2001 From: Sebastian Borza Date: Wed, 8 Apr 2020 12:39:13 -0500 Subject: [PATCH 03/37] feat(line_protocol): split out make_line function from core make_lines (#810) * feat(line_protocol): split out make_line function from core make_lines * chore(line_protocol): fix malformed testcase --- CHANGELOG.md | 1 + influxdb/line_protocol.py | 123 ++++++++++++++++----------- influxdb/tests/test_line_protocol.py | 2 +- 3 files changed, 75 insertions(+), 51 deletions(-) diff --git a/CHANGELOG.md b/CHANGELOG.md index 59777b5f..81809bcd 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -21,6 +21,7 @@ and this project adheres to [Semantic Versioning](http://semver.org/spec/v2.0.0. ### Changed - Clean up stale CI config (#755) - Add legacy client test (#752 & #318 thx @oldmantaiter & @sebito91) +- Update make_lines section in line_protocol.py to split out core function (#375 thx @aisbaa) ### Removed diff --git a/influxdb/line_protocol.py b/influxdb/line_protocol.py index 249511d3..ec59ef47 100644 --- a/influxdb/line_protocol.py +++ b/influxdb/line_protocol.py @@ -11,7 +11,7 @@ from pytz import UTC from dateutil.parser import parse -from six import iteritems, binary_type, text_type, integer_types, PY2 +from six import binary_type, text_type, integer_types, PY2 EPOCH = UTC.localize(datetime.utcfromtimestamp(0)) @@ -30,15 +30,20 @@ def _convert_timestamp(timestamp, precision=None): ns = (timestamp - EPOCH).total_seconds() * 1e9 if precision is None or precision == 'n': return ns - elif precision == 'u': + + if precision == 'u': return ns / 1e3 - elif precision == 'ms': + + if precision == 'ms': return ns / 1e6 - elif precision == 's': + + if precision == 's': return ns / 1e9 - elif precision == 'm': + + if precision == 'm': return ns / 1e9 / 60 - elif precision == 'h': + + if precision == 'h': return ns / 1e9 / 3600 raise ValueError(timestamp) @@ -95,9 +100,11 @@ def _escape_value(value): if isinstance(value, text_type) and value != '': return quote_ident(value) - elif isinstance(value, integer_types) and not isinstance(value, bool): + + if isinstance(value, integer_types) and not isinstance(value, bool): return str(value) + 'i' - elif _is_float(value): + + if _is_float(value): return repr(value) return str(value) @@ -107,15 +114,60 @@ def _get_unicode(data, force=False): """Try to return a text aka unicode object from the given data.""" if isinstance(data, binary_type): return data.decode('utf-8') - elif data is None: + + if data is None: return '' - elif force: + + if force: if PY2: return unicode(data) - else: - return str(data) - else: - return data + return str(data) + + return data + + +def make_line(measurement, tags=None, fields=None, time=None, precision=None): + """Extract the actual point from a given measurement line.""" + tags = tags or {} + fields = fields or {} + + line = _escape_tag(_get_unicode(measurement)) + + # tags should be sorted client-side to take load off server + tag_list = [] + for tag_key in sorted(tags.keys()): + key = _escape_tag(tag_key) + value = _escape_tag(tags[tag_key]) + + if key != '' and value != '': + tag_list.append( + "{key}={value}".format(key=key, value=value) + ) + + if tag_list: + line += ',' + ','.join(tag_list) + + field_list = [] + for field_key in sorted(fields.keys()): + key = _escape_tag(field_key) + value = _escape_value(fields[field_key]) + + if key != '' and value != '': + field_list.append("{key}={value}".format( + key=key, + value=value + )) + + if field_list: + line += ' ' + ','.join(field_list) + + if time is not None: + timestamp = _get_unicode(str(int( + _convert_timestamp(time, precision) + ))) + line += ' ' + timestamp + + return line def make_lines(data, precision=None): @@ -127,48 +179,19 @@ def make_lines(data, precision=None): lines = [] static_tags = data.get('tags') for point in data['points']: - elements = [] - - # add measurement name - measurement = _escape_tag(_get_unicode( - point.get('measurement', data.get('measurement')))) - key_values = [measurement] - - # add tags if static_tags: tags = dict(static_tags) # make a copy, since we'll modify tags.update(point.get('tags') or {}) else: tags = point.get('tags') or {} - # tags should be sorted client-side to take load off server - for tag_key, tag_value in sorted(iteritems(tags)): - key = _escape_tag(tag_key) - value = _escape_tag_value(tag_value) - - if key != '' and value != '': - key_values.append(key + "=" + value) - - elements.append(','.join(key_values)) - - # add fields - field_values = [] - for field_key, field_value in sorted(iteritems(point['fields'])): - key = _escape_tag(field_key) - value = _escape_value(field_value) - - if key != '' and value != '': - field_values.append(key + "=" + value) - - elements.append(','.join(field_values)) - - # add timestamp - if 'time' in point: - timestamp = _get_unicode(str(int( - _convert_timestamp(point['time'], precision)))) - elements.append(timestamp) - - line = ' '.join(elements) + line = make_line( + point.get('measurement', data.get('measurement')), + tags=tags, + fields=point.get('fields'), + precision=precision, + time=point.get('time') + ) lines.append(line) return '\n'.join(lines) + '\n' diff --git a/influxdb/tests/test_line_protocol.py b/influxdb/tests/test_line_protocol.py index bccd7727..71828f62 100644 --- a/influxdb/tests/test_line_protocol.py +++ b/influxdb/tests/test_line_protocol.py @@ -42,7 +42,7 @@ def test_make_lines(self): self.assertEqual( line_protocol.make_lines(data), - 'test,backslash_tag=C:\\\\ ,integer_tag=2,string_tag=hello ' + 'test,backslash_tag=C:\\\\,integer_tag=2,string_tag=hello ' 'bool_val=True,float_val=1.1,int_val=1i,string_val="hello!"\n' ) From 62436d4583b6a6d0ea002e6d490b2afa1e410956 Mon Sep 17 00:00:00 2001 From: Sebastian Borza Date: Wed, 8 Apr 2020 16:09:04 -0500 Subject: [PATCH 04/37] chore(line_protocol): fix nanosecond timestamp resolution for points (#811) --- CHANGELOG.md | 1 + influxdb/line_protocol.py | 20 ++++++++++++++------ 2 files changed, 15 insertions(+), 6 deletions(-) diff --git a/CHANGELOG.md b/CHANGELOG.md index 81809bcd..0556d942 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -22,6 +22,7 @@ and this project adheres to [Semantic Versioning](http://semver.org/spec/v2.0.0. - Clean up stale CI config (#755) - Add legacy client test (#752 & #318 thx @oldmantaiter & @sebito91) - Update make_lines section in line_protocol.py to split out core function (#375 thx @aisbaa) +- Fix nanosecond time resolution for points (#407 thx @AndreCAndersen && @clslgrnc) ### Removed diff --git a/influxdb/line_protocol.py b/influxdb/line_protocol.py index ec59ef47..3a5eb4e8 100644 --- a/influxdb/line_protocol.py +++ b/influxdb/line_protocol.py @@ -16,6 +16,14 @@ EPOCH = UTC.localize(datetime.utcfromtimestamp(0)) +def _to_nanos(timestamp): + delta = timestamp - EPOCH + nanos_in_days = delta.days * 86400 * 10 ** 9 + nanos_in_seconds = delta.seconds * 10 ** 9 + nanos_in_micros = delta.microseconds * 10 ** 3 + return nanos_in_days + nanos_in_seconds + nanos_in_micros + + def _convert_timestamp(timestamp, precision=None): if isinstance(timestamp, Integral): return timestamp # assume precision is correct if timestamp is int @@ -27,24 +35,24 @@ def _convert_timestamp(timestamp, precision=None): if not timestamp.tzinfo: timestamp = UTC.localize(timestamp) - ns = (timestamp - EPOCH).total_seconds() * 1e9 + ns = _to_nanos(timestamp) if precision is None or precision == 'n': return ns if precision == 'u': - return ns / 1e3 + return ns / 10**3 if precision == 'ms': - return ns / 1e6 + return ns / 10**6 if precision == 's': - return ns / 1e9 + return ns / 10**9 if precision == 'm': - return ns / 1e9 / 60 + return ns / 10**9 / 60 if precision == 'h': - return ns / 1e9 / 3600 + return ns / 10**9 / 3600 raise ValueError(timestamp) From 9382f94785e449eaf624ae3b5fa39baf47456af8 Mon Sep 17 00:00:00 2001 From: Sergei Smolianinov Date: Wed, 8 Apr 2020 21:21:12 +0000 Subject: [PATCH 05/37] Add CQs management methods to the client (#414) * Add CQs management methods to the client * chore(server_tests): update pep257 and flake8 commentary * chore(client_test): update comments based on pep257 and flake8 Co-authored-by: Sergei Smolianinov Co-authored-by: Sebastian Borza Co-authored-by: xginn8 --- influxdb/tests/client_test.py | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/influxdb/tests/client_test.py b/influxdb/tests/client_test.py index 99a7f42b..fe7381ea 100644 --- a/influxdb/tests/client_test.py +++ b/influxdb/tests/client_test.py @@ -1161,7 +1161,7 @@ def test_revoke_privilege_invalid(self): self.cli.revoke_privilege('', 'testdb', 'test') def test_get_list_privileges(self): - """Tst get list of privs for TestInfluxDBClient object.""" + """Test get list of privs for TestInfluxDBClient object.""" data = {'results': [ {'series': [ {'columns': ['database', 'privilege'], From fb1441a9fb1b799ef5cb42a80562a0a7706463e6 Mon Sep 17 00:00:00 2001 From: Sebastian Borza Date: Wed, 8 Apr 2020 16:22:27 -0500 Subject: [PATCH 06/37] chore(CHANGELOG): add smolse to PR merge for #681 and #414 --- CHANGELOG.md | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/CHANGELOG.md b/CHANGELOG.md index 0556d942..2a887f90 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -34,7 +34,7 @@ and this project adheres to [Semantic Versioning](http://semver.org/spec/v2.0.0. - Add consistency paramter to `write_points` (#664 tx @RonRothman) - The query() function now accepts a bind_params argument for parameter binding (#678 thx @clslgrnc) - Add `get_list_continuous_queries`, `drop_continuous_query`, and `create_continuous_query` management methods for - continuous queries (#681 thx @lukaszdudek-silvair) + continuous queries (#681 thx @lukaszdudek-silvair && @smolse) - Mutual TLS authentication (#702 thx @LloydW93) ### Changed From e0da05bb8e3815fc81129ac9b15bb6a2a47276d3 Mon Sep 17 00:00:00 2001 From: Sebastian Borza Date: Wed, 8 Apr 2020 19:22:31 -0500 Subject: [PATCH 07/37] feat(dataframe_client): handle np.nan, np.inf values in DataFrameClient (#812) * feat(dataframe_client): handle np.nan, np.inf values in DataFrameClient * chore(dataframe): handle cases where tagset is empty * chore(dataframe): add tests for Nan lines but with tag values --- CHANGELOG.md | 1 + influxdb/_dataframe_client.py | 40 +++++--- influxdb/tests/dataframe_client_test.py | 122 +++++++++++++++++++++++- 3 files changed, 147 insertions(+), 16 deletions(-) diff --git a/CHANGELOG.md b/CHANGELOG.md index 2a887f90..be15d52b 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -17,6 +17,7 @@ and this project adheres to [Semantic Versioning](http://semver.org/spec/v2.0.0. - Add support for messagepack (#734 thx @lovasoa) - Add support for 'show series' (#357 thx @gaker) - Add support for custom request session in InfluxDBClient (#360 thx @dschien) +- Add support for handling np.nan and np.inf values in DataFrameClient (#436 thx @nmerket) ### Changed - Clean up stale CI config (#755) diff --git a/influxdb/_dataframe_client.py b/influxdb/_dataframe_client.py index d16e29ca..f411bb37 100644 --- a/influxdb/_dataframe_client.py +++ b/influxdb/_dataframe_client.py @@ -270,14 +270,31 @@ def _convert_dataframe_to_json(dataframe, "h": 1e9 * 3600, }.get(time_precision, 1) + if not tag_columns: + points = [ + {'measurement': measurement, + 'fields': + rec.replace([np.inf, -np.inf], np.nan).dropna().to_dict(), + 'time': np.int64(ts.value / precision_factor)} + for ts, (_, rec) in zip( + dataframe.index, + dataframe[field_columns].iterrows() + ) + ] + + return points + points = [ {'measurement': measurement, 'tags': dict(list(tag.items()) + list(tags.items())), - 'fields': rec, + 'fields': + rec.replace([np.inf, -np.inf], np.nan).dropna().to_dict(), 'time': np.int64(ts.value / precision_factor)} - for ts, tag, rec in zip(dataframe.index, - dataframe[tag_columns].to_dict('record'), - dataframe[field_columns].to_dict('record')) + for ts, tag, (_, rec) in zip( + dataframe.index, + dataframe[tag_columns].to_dict('record'), + dataframe[field_columns].iterrows() + ) ] return points @@ -379,21 +396,18 @@ def _convert_dataframe_to_lines(self, tags = '' # Make an array of formatted field keys and values - field_df = dataframe[field_columns] - # Keep the positions where Null values are found - mask_null = field_df.isnull().values + field_df = dataframe[field_columns].replace([np.inf, -np.inf], np.nan) + nans = pd.isnull(field_df) field_df = self._stringify_dataframe(field_df, numeric_precision, datatype='field') field_df = (field_df.columns.values + '=').tolist() + field_df - field_df[field_df.columns[1:]] = ',' + field_df[ - field_df.columns[1:]] - field_df = field_df.where(~mask_null, '') # drop Null entries - fields = field_df.sum(axis=1) - # take out leading , where first column has a Null value - fields = fields.str.lstrip(",") + field_df[field_df.columns[1:]] = ',' + field_df[field_df.columns[1:]] + field_df[nans] = '' + + fields = field_df.sum(axis=1).map(lambda x: x.lstrip(',')) del field_df # Generate line protocol string diff --git a/influxdb/tests/dataframe_client_test.py b/influxdb/tests/dataframe_client_test.py index 90312ed8..0573d5c3 100644 --- a/influxdb/tests/dataframe_client_test.py +++ b/influxdb/tests/dataframe_client_test.py @@ -13,8 +13,8 @@ import warnings import requests_mock -from influxdb.tests import skip_if_pypy, using_pypy from nose.tools import raises +from influxdb.tests import skip_if_pypy, using_pypy from .client_test import _mocked_session @@ -22,7 +22,7 @@ import pandas as pd from pandas.util.testing import assert_frame_equal from influxdb import DataFrameClient - import numpy + import numpy as np @skip_if_pypy @@ -462,7 +462,7 @@ def test_write_points_from_dataframe_with_numeric_precision(self): ["2", 2, 2.2222222222222]], index=[now, now + timedelta(hours=1)]) - if numpy.lib.NumpyVersion(numpy.__version__) <= '1.13.3': + if np.lib.NumpyVersion(np.__version__) <= '1.13.3': expected_default_precision = ( b'foo,hello=there 0=\"1\",1=1i,2=1.11111111111 0\n' b'foo,hello=there 0=\"2\",1=2i,2=2.22222222222 3600000000000\n' @@ -1032,3 +1032,119 @@ def test_dsn_constructor(self): client = DataFrameClient.from_dsn('influxdb://localhost:8086') self.assertIsInstance(client, DataFrameClient) self.assertEqual('http://localhost:8086', client._baseurl) + + def test_write_points_from_dataframe_with_nan_line(self): + """Test write points from dataframe with Nan lines.""" + now = pd.Timestamp('1970-01-01 00:00+00:00') + dataframe = pd.DataFrame(data=[["1", 1, np.inf], ["2", 2, np.nan]], + index=[now, now + timedelta(hours=1)], + columns=["column_one", "column_two", + "column_three"]) + expected = ( + b"foo column_one=\"1\",column_two=1i 0\n" + b"foo column_one=\"2\",column_two=2i " + b"3600000000000\n" + ) + + with requests_mock.Mocker() as m: + m.register_uri(requests_mock.POST, + "http://localhost:8086/write", + status_code=204) + + cli = DataFrameClient(database='db') + + cli.write_points(dataframe, 'foo', protocol='line') + self.assertEqual(m.last_request.body, expected) + + cli.write_points(dataframe, 'foo', tags=None, protocol='line') + self.assertEqual(m.last_request.body, expected) + + def test_write_points_from_dataframe_with_nan_json(self): + """Test write points from json with NaN lines.""" + now = pd.Timestamp('1970-01-01 00:00+00:00') + dataframe = pd.DataFrame(data=[["1", 1, np.inf], ["2", 2, np.nan]], + index=[now, now + timedelta(hours=1)], + columns=["column_one", "column_two", + "column_three"]) + expected = ( + b"foo column_one=\"1\",column_two=1i 0\n" + b"foo column_one=\"2\",column_two=2i " + b"3600000000000\n" + ) + + with requests_mock.Mocker() as m: + m.register_uri(requests_mock.POST, + "http://localhost:8086/write", + status_code=204) + + cli = DataFrameClient(database='db') + + cli.write_points(dataframe, 'foo', protocol='json') + self.assertEqual(m.last_request.body, expected) + + cli.write_points(dataframe, 'foo', tags=None, protocol='json') + self.assertEqual(m.last_request.body, expected) + + def test_write_points_from_dataframe_with_tags_and_nan_line(self): + """Test write points from dataframe with NaN lines and tags.""" + now = pd.Timestamp('1970-01-01 00:00+00:00') + dataframe = pd.DataFrame(data=[['blue', 1, "1", 1, np.inf], + ['red', 0, "2", 2, np.nan]], + index=[now, now + timedelta(hours=1)], + columns=["tag_one", "tag_two", "column_one", + "column_two", "column_three"]) + expected = ( + b"foo,tag_one=blue,tag_two=1 " + b"column_one=\"1\",column_two=1i " + b"0\n" + b"foo,tag_one=red,tag_two=0 " + b"column_one=\"2\",column_two=2i " + b"3600000000000\n" + ) + + with requests_mock.Mocker() as m: + m.register_uri(requests_mock.POST, + "http://localhost:8086/write", + status_code=204) + + cli = DataFrameClient(database='db') + + cli.write_points(dataframe, 'foo', protocol='line', + tag_columns=['tag_one', 'tag_two']) + self.assertEqual(m.last_request.body, expected) + + cli.write_points(dataframe, 'foo', tags=None, protocol='line', + tag_columns=['tag_one', 'tag_two']) + self.assertEqual(m.last_request.body, expected) + + def test_write_points_from_dataframe_with_tags_and_nan_json(self): + """Test write points from json with NaN lines and tags.""" + now = pd.Timestamp('1970-01-01 00:00+00:00') + dataframe = pd.DataFrame(data=[['blue', 1, "1", 1, np.inf], + ['red', 0, "2", 2, np.nan]], + index=[now, now + timedelta(hours=1)], + columns=["tag_one", "tag_two", "column_one", + "column_two", "column_three"]) + expected = ( + b"foo,tag_one=blue,tag_two=1 " + b"column_one=\"1\",column_two=1i " + b"0\n" + b"foo,tag_one=red,tag_two=0 " + b"column_one=\"2\",column_two=2i " + b"3600000000000\n" + ) + + with requests_mock.Mocker() as m: + m.register_uri(requests_mock.POST, + "http://localhost:8086/write", + status_code=204) + + cli = DataFrameClient(database='db') + + cli.write_points(dataframe, 'foo', protocol='json', + tag_columns=['tag_one', 'tag_two']) + self.assertEqual(m.last_request.body, expected) + + cli.write_points(dataframe, 'foo', tags=None, protocol='json', + tag_columns=['tag_one', 'tag_two']) + self.assertEqual(m.last_request.body, expected) From aa46848ec6bf41d3c8de1d3857e3a7e670b8f005 Mon Sep 17 00:00:00 2001 From: Christopher Head Date: Thu, 9 Apr 2020 15:33:11 -0700 Subject: [PATCH 08/37] Fix import of distutils.spawn (#805) --- influxdb/tests/server_tests/influxdb_instance.py | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/influxdb/tests/server_tests/influxdb_instance.py b/influxdb/tests/server_tests/influxdb_instance.py index 1dcd7567..2dd823ff 100644 --- a/influxdb/tests/server_tests/influxdb_instance.py +++ b/influxdb/tests/server_tests/influxdb_instance.py @@ -7,7 +7,7 @@ from __future__ import unicode_literals import datetime -import distutils +import distutils.spawn import os import tempfile import shutil From 952e834cd3d7996f719512832a00f6648e07daca Mon Sep 17 00:00:00 2001 From: Sebastian Borza Date: Thu, 9 Apr 2020 17:34:33 -0500 Subject: [PATCH 09/37] chore(CHANGELOG): update to include PR from #805 --- CHANGELOG.md | 3 ++- 1 file changed, 2 insertions(+), 1 deletion(-) diff --git a/CHANGELOG.md b/CHANGELOG.md index be15d52b..44f7f4d1 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -10,7 +10,7 @@ and this project adheres to [Semantic Versioning](http://semver.org/spec/v2.0.0. ### Changed -## [v5.2.4] - 2020-04-10 +## [v5.3.0] - 2020-04-10 ### Added - Add mypy testing framework (#756) @@ -24,6 +24,7 @@ and this project adheres to [Semantic Versioning](http://semver.org/spec/v2.0.0. - Add legacy client test (#752 & #318 thx @oldmantaiter & @sebito91) - Update make_lines section in line_protocol.py to split out core function (#375 thx @aisbaa) - Fix nanosecond time resolution for points (#407 thx @AndreCAndersen && @clslgrnc) +- Fix import of distutils.spawn (#805 thx @Hawk777) ### Removed From cd5999e53a9bbb2e50db463e796ce2cfc1d7dc53 Mon Sep 17 00:00:00 2001 From: Sebastian Borza Date: Thu, 9 Apr 2020 19:05:34 -0500 Subject: [PATCH 10/37] chore(line_protocol): update repr value of floats to properly handle precision (#813) * chore(line_protocol): update repr value of floats to properly handle precision. Closes #488 * chore(line_protocol): fix repr and handle boolean values * chore(CHANGELOG): update to include reference to PR#488 --- CHANGELOG.md | 1 + influxdb/line_protocol.py | 5 ++++- influxdb/tests/test_line_protocol.py | 23 +++++++++++++++++++++-- 3 files changed, 26 insertions(+), 3 deletions(-) diff --git a/CHANGELOG.md b/CHANGELOG.md index 44f7f4d1..943fb83f 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -25,6 +25,7 @@ and this project adheres to [Semantic Versioning](http://semver.org/spec/v2.0.0. - Update make_lines section in line_protocol.py to split out core function (#375 thx @aisbaa) - Fix nanosecond time resolution for points (#407 thx @AndreCAndersen && @clslgrnc) - Fix import of distutils.spawn (#805 thx @Hawk777) +- Update repr of float values including properly handling of boolean (#488 thx @ghost) ### Removed diff --git a/influxdb/line_protocol.py b/influxdb/line_protocol.py index 3a5eb4e8..d6cbf46f 100644 --- a/influxdb/line_protocol.py +++ b/influxdb/line_protocol.py @@ -112,8 +112,11 @@ def _escape_value(value): if isinstance(value, integer_types) and not isinstance(value, bool): return str(value) + 'i' + if isinstance(value, bool): + return str(value) + if _is_float(value): - return repr(value) + return repr(float(value)) return str(value) diff --git a/influxdb/tests/test_line_protocol.py b/influxdb/tests/test_line_protocol.py index 71828f62..c48d5edc 100644 --- a/influxdb/tests/test_line_protocol.py +++ b/influxdb/tests/test_line_protocol.py @@ -6,10 +6,12 @@ from __future__ import print_function from __future__ import unicode_literals -from datetime import datetime import unittest -from pytz import UTC, timezone +from datetime import datetime +from decimal import Decimal + +from pytz import UTC, timezone from influxdb import line_protocol @@ -166,3 +168,20 @@ def test_float_with_long_decimal_fraction(self): line_protocol.make_lines(data), 'test float_val=1.0000000000000009\n' ) + + def test_float_with_long_decimal_fraction_as_type_decimal(self): + """Ensure precision is preserved when casting Decimal into strings.""" + data = { + "points": [ + { + "measurement": "test", + "fields": { + "float_val": Decimal(0.8289445733333332), + } + } + ] + } + self.assertEqual( + line_protocol.make_lines(data), + 'test float_val=0.8289445733333332\n' + ) From 4a92377e556ad5e051342f1321ab3fb6a9d7572d Mon Sep 17 00:00:00 2001 From: Kenzyme Le Date: Thu, 9 Apr 2020 22:15:05 -0400 Subject: [PATCH 11/37] Fix tests for existing 'Adding time_precision optional option to SeriesHelper' PR (#719) * Adding time_precision into Meta of SeriesHelper time_precision option not currently supported in influx-db python * Making Time Precision optional Changed it from required to optional * Fixing Typo in _time_precision Attribute * Fixing coding conventions for Travis CI * Appunni: Test for invalid time_precision on SeriesHelper * Appunni: Intendation problem resolution in master * - Fix flake7 errors : E131 continuation line unaligned for hanging indent - Fix typo - Fix cls._client declaration ordering - Remove duplicate code cls._autocommit ... Co-authored-by: appunni-dishq <31534711+appunni-dishq@users.noreply.github.com> Co-authored-by: xginn8 Co-authored-by: appunni --- influxdb/helper.py | 20 +++++++++++++++++--- influxdb/tests/helper_test.py | 13 ++++++++++++- 2 files changed, 29 insertions(+), 4 deletions(-) diff --git a/influxdb/helper.py b/influxdb/helper.py index e622526d..fa79c079 100644 --- a/influxdb/helper.py +++ b/influxdb/helper.py @@ -41,6 +41,10 @@ class Meta: # Only applicable if autocommit is True. autocommit = True # If True and no bulk_size, then will set bulk_size to 1. + time_precision = "h"|"m"|s"|"ms"|"u"|"ns" + # Default is ns (nanoseconds) + # Setting time precision while writing point + # You should also make sure time is set in the given precision """ @@ -71,6 +75,13 @@ def __new__(cls, *args, **kwargs): cls.__name__)) cls._autocommit = getattr(_meta, 'autocommit', False) + cls._time_precision = getattr(_meta, 'time_precision', None) + + allowed_time_precisions = ['h', 'm', 's', 'ms', 'u', 'ns', None] + if cls._time_precision not in allowed_time_precisions: + raise AttributeError( + 'In {0}, time_precision is set, but invalid use any of {}.' + .format(cls.__name__, ','.join(allowed_time_precisions))) cls._client = getattr(_meta, 'client', None) if cls._autocommit and not cls._client: @@ -116,11 +127,11 @@ def __init__(self, **kw): keys = set(kw.keys()) # all tags should be passed, and keys - tags should be a subset of keys - if not(tags <= keys): + if not (tags <= keys): raise NameError( 'Expected arguments to contain all tags {0}, instead got {1}.' .format(cls._tags, kw.keys())) - if not(keys - tags <= fields): + if not (keys - tags <= fields): raise NameError('Got arguments not in tags or fields: {0}' .format(keys - tags - fields)) @@ -143,7 +154,10 @@ def commit(cls, client=None): """ if not client: client = cls._client - rtn = client.write_points(cls._json_body_()) + rtn = client.write_points( + cls._json_body_(), + time_precision=cls._time_precision) + # will be None if not set and will default to ns cls._reset_() return rtn diff --git a/influxdb/tests/helper_test.py b/influxdb/tests/helper_test.py index 6f24e85d..6aa8f15a 100644 --- a/influxdb/tests/helper_test.py +++ b/influxdb/tests/helper_test.py @@ -310,8 +310,19 @@ class Meta: series_name = 'events.stats.{server_name}' + class InvalidTimePrecision(SeriesHelper): + """Define instance of SeriesHelper for invalid time precision.""" + + class Meta: + """Define metadata for InvalidTimePrecision.""" + + series_name = 'events.stats.{server_name}' + time_precision = "ks" + fields = ['time', 'server_name'] + autocommit = True + for cls in [MissingMeta, MissingClient, MissingFields, - MissingSeriesName]: + MissingSeriesName, InvalidTimePrecision]: self.assertRaises( AttributeError, cls, **{'time': 159, 'server_name': 'us.east-1'}) From 774a14017669040ffe4fcffa9eb441e9a07502a9 Mon Sep 17 00:00:00 2001 From: Sebastian Borza Date: Thu, 9 Apr 2020 21:27:23 -0500 Subject: [PATCH 12/37] chore(CHANGELOG): update for PR#502 and #719 --- CHANGELOG.md | 1 + 1 file changed, 1 insertion(+) diff --git a/CHANGELOG.md b/CHANGELOG.md index 943fb83f..b94e438e 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -18,6 +18,7 @@ and this project adheres to [Semantic Versioning](http://semver.org/spec/v2.0.0. - Add support for 'show series' (#357 thx @gaker) - Add support for custom request session in InfluxDBClient (#360 thx @dschien) - Add support for handling np.nan and np.inf values in DataFrameClient (#436 thx @nmerket) +- Add support for optional `time_precision` in the SeriesHelper (#502 && #719 thx @appunni-dishq && @klDen) ### Changed - Clean up stale CI config (#755) From 2e11b95fe222727c666e488d7147eb878acf1797 Mon Sep 17 00:00:00 2001 From: Sebastian Borza Date: Fri, 10 Apr 2020 08:59:46 -0500 Subject: [PATCH 13/37] chore(dataframe_client): update to handle empty tags in dataframe client (#814) --- CHANGELOG.md | 1 + influxdb/_dataframe_client.py | 3 ++- 2 files changed, 3 insertions(+), 1 deletion(-) diff --git a/CHANGELOG.md b/CHANGELOG.md index b94e438e..e7f40dc7 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -27,6 +27,7 @@ and this project adheres to [Semantic Versioning](http://semver.org/spec/v2.0.0. - Fix nanosecond time resolution for points (#407 thx @AndreCAndersen && @clslgrnc) - Fix import of distutils.spawn (#805 thx @Hawk777) - Update repr of float values including properly handling of boolean (#488 thx @ghost) +- Update dataframe_client to fix faulty empty tags (#770 thx @michelfripiat) ### Removed diff --git a/influxdb/_dataframe_client.py b/influxdb/_dataframe_client.py index f411bb37..a977754e 100644 --- a/influxdb/_dataframe_client.py +++ b/influxdb/_dataframe_client.py @@ -388,7 +388,8 @@ def _convert_dataframe_to_lines(self, del tag_df elif global_tags: tag_string = ''.join( - [",{}={}".format(k, _escape_tag(v)) if v else '' + [",{}={}".format(k, _escape_tag(v)) + if v not in [None, ''] else "" for k, v in sorted(global_tags.items())] ) tags = pd.Series(tag_string, index=dataframe.index) From d02123548abc3c0a8b1fc8ac7e29f0f7977a148b Mon Sep 17 00:00:00 2001 From: jgspiro <45625897+jgspiro@users.noreply.github.com> Date: Fri, 10 Apr 2020 16:07:18 +0200 Subject: [PATCH 14/37] Bugfix dropna in DataFrameClient. Add unit test. (#778) --- influxdb/_dataframe_client.py | 3 +- influxdb/tests/dataframe_client_test.py | 92 +++++++++++++++++++++++++ 2 files changed, 94 insertions(+), 1 deletion(-) diff --git a/influxdb/_dataframe_client.py b/influxdb/_dataframe_client.py index a977754e..d7e67baa 100644 --- a/influxdb/_dataframe_client.py +++ b/influxdb/_dataframe_client.py @@ -203,7 +203,8 @@ def query(self, def _to_dataframe(self, rs, dropna=True): result = defaultdict(list) if isinstance(rs, list): - return map(self._to_dataframe, rs) + return map(self._to_dataframe, rs, + [dropna for _ in range(len(rs))]) for key, data in rs.items(): name, tags = key diff --git a/influxdb/tests/dataframe_client_test.py b/influxdb/tests/dataframe_client_test.py index 0573d5c3..4e172ea7 100644 --- a/influxdb/tests/dataframe_client_test.py +++ b/influxdb/tests/dataframe_client_test.py @@ -968,6 +968,98 @@ def test_multiquery_into_dataframe(self): for k in e: assert_frame_equal(e[k], r[k]) + def test_multiquery_into_dataframe_dropna(self): + """Test multiquery into df for TestDataFrameClient object.""" + data = { + "results": [ + { + "series": [ + { + "name": "cpu_load_short", + "columns": ["time", "value", "value2", "value3"], + "values": [ + ["2015-01-29T21:55:43.702900257Z", + 0.55, 0.254, numpy.NaN], + ["2015-01-29T21:55:43.702900257Z", + 23422, 122878, numpy.NaN], + ["2015-06-11T20:46:02Z", + 0.64, 0.5434, numpy.NaN] + ] + } + ] + }, { + "series": [ + { + "name": "cpu_load_short", + "columns": ["time", "count"], + "values": [ + ["1970-01-01T00:00:00Z", 3] + ] + } + ] + } + ] + } + + pd1 = pd.DataFrame( + [[0.55, 0.254, numpy.NaN], + [23422.0, 122878, numpy.NaN], + [0.64, 0.5434, numpy.NaN]], + columns=['value', 'value2', 'value3'], + index=pd.to_datetime([ + "2015-01-29 21:55:43.702900257+0000", + "2015-01-29 21:55:43.702900257+0000", + "2015-06-11 20:46:02+0000"])) + + if pd1.index.tzinfo is None: + pd1.index = pd1.index.tz_localize('UTC') + + pd1_dropna = pd.DataFrame( + [[0.55, 0.254], [23422.0, 122878], [0.64, 0.5434]], + columns=['value', 'value2'], + index=pd.to_datetime([ + "2015-01-29 21:55:43.702900257+0000", + "2015-01-29 21:55:43.702900257+0000", + "2015-06-11 20:46:02+0000"])) + + if pd1_dropna.index.tzinfo is None: + pd1_dropna.index = pd1_dropna.index.tz_localize('UTC') + + pd2 = pd.DataFrame( + [[3]], columns=['count'], + index=pd.to_datetime(["1970-01-01 00:00:00+00:00"])) + + if pd2.index.tzinfo is None: + pd2.index = pd2.index.tz_localize('UTC') + + expected_dropna_true = [ + {'cpu_load_short': pd1_dropna}, + {'cpu_load_short': pd2}] + expected_dropna_false = [ + {'cpu_load_short': pd1}, + {'cpu_load_short': pd2}] + + cli = DataFrameClient('host', 8086, 'username', 'password', 'db') + iql = "SELECT value FROM cpu_load_short WHERE region=$region;" \ + "SELECT count(value) FROM cpu_load_short WHERE region=$region" + bind_params = {'region': 'us-west'} + + for dropna in [True, False]: + with _mocked_session(cli, 'GET', 200, data): + result = cli.query(iql, bind_params=bind_params, dropna=dropna) + expected = \ + expected_dropna_true if dropna else expected_dropna_false + for r, e in zip(result, expected): + for k in e: + assert_frame_equal(e[k], r[k]) + + # test default value (dropna = True) + with _mocked_session(cli, 'GET', 200, data): + result = cli.query(iql, bind_params=bind_params) + for r, e in zip(result, expected_dropna_true): + for k in e: + assert_frame_equal(e[k], r[k]) + def test_query_with_empty_result(self): """Test query with empty results in TestDataFrameClient object.""" cli = DataFrameClient('host', 8086, 'username', 'password', 'db') From cb31bf58322cbcd34d52792330b5c1418db388d9 Mon Sep 17 00:00:00 2001 From: testforvln <1694611+testforvln@users.noreply.github.com> Date: Fri, 10 Apr 2020 22:21:24 +0800 Subject: [PATCH 15/37] fix bug in _convert_dataframe_to_json function that need not transform index if data type of index is already DatatimeIndex (#623) --- influxdb/_dataframe_client.py | 3 ++- 1 file changed, 2 insertions(+), 1 deletion(-) diff --git a/influxdb/_dataframe_client.py b/influxdb/_dataframe_client.py index d7e67baa..600bc1ec 100644 --- a/influxdb/_dataframe_client.py +++ b/influxdb/_dataframe_client.py @@ -252,7 +252,8 @@ def _convert_dataframe_to_json(dataframe, field_columns = list( set(dataframe.columns).difference(set(tag_columns))) - dataframe.index = pd.to_datetime(dataframe.index) + if not isinstance(dataframe.index, pd.DatetimeIndex): + dataframe.index = pd.to_datetime(dataframe.index) if dataframe.index.tzinfo is None: dataframe.index = dataframe.index.tz_localize('UTC') From cc903328ca463ee6554fb9beaed5a1dfad304d74 Mon Sep 17 00:00:00 2001 From: Sebastian Borza Date: Fri, 10 Apr 2020 09:22:36 -0500 Subject: [PATCH 16/37] chore(CHANGELOG): update with more recent PR merges --- CHANGELOG.md | 4 +++- 1 file changed, 3 insertions(+), 1 deletion(-) diff --git a/CHANGELOG.md b/CHANGELOG.md index e7f40dc7..11f47e38 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -27,7 +27,9 @@ and this project adheres to [Semantic Versioning](http://semver.org/spec/v2.0.0. - Fix nanosecond time resolution for points (#407 thx @AndreCAndersen && @clslgrnc) - Fix import of distutils.spawn (#805 thx @Hawk777) - Update repr of float values including properly handling of boolean (#488 thx @ghost) -- Update dataframe_client to fix faulty empty tags (#770 thx @michelfripiat) +- Update DataFrameClient to fix faulty empty tags (#770 thx @michelfripiat) +- Update DataFrameClient to properly return `dropna` values (#778 thx @jgspiro) +- Update DataFrameClient to test for pd.DataTimeIndex before blind conversion (#623 thx @testforvin) ### Removed From f35e3315b8f8c37d51cb23f56522d7b7ceb7bf82 Mon Sep 17 00:00:00 2001 From: Sebastian Borza Date: Fri, 10 Apr 2020 09:41:52 -0500 Subject: [PATCH 17/37] chore(dataframe_client_test): fix incorrect package naming convention --- influxdb/tests/dataframe_client_test.py | 12 ++++++------ 1 file changed, 6 insertions(+), 6 deletions(-) diff --git a/influxdb/tests/dataframe_client_test.py b/influxdb/tests/dataframe_client_test.py index 4e172ea7..a80498f3 100644 --- a/influxdb/tests/dataframe_client_test.py +++ b/influxdb/tests/dataframe_client_test.py @@ -979,11 +979,11 @@ def test_multiquery_into_dataframe_dropna(self): "columns": ["time", "value", "value2", "value3"], "values": [ ["2015-01-29T21:55:43.702900257Z", - 0.55, 0.254, numpy.NaN], + 0.55, 0.254, np.NaN], ["2015-01-29T21:55:43.702900257Z", - 23422, 122878, numpy.NaN], + 23422, 122878, np.NaN], ["2015-06-11T20:46:02Z", - 0.64, 0.5434, numpy.NaN] + 0.64, 0.5434, np.NaN] ] } ] @@ -1002,9 +1002,9 @@ def test_multiquery_into_dataframe_dropna(self): } pd1 = pd.DataFrame( - [[0.55, 0.254, numpy.NaN], - [23422.0, 122878, numpy.NaN], - [0.64, 0.5434, numpy.NaN]], + [[0.55, 0.254, np.NaN], + [23422.0, 122878, np.NaN], + [0.64, 0.5434, np.NaN]], columns=['value', 'value2', 'value3'], index=pd.to_datetime([ "2015-01-29 21:55:43.702900257+0000", From 463f221269afb9fadac3d6b50ea92219e17e4df8 Mon Sep 17 00:00:00 2001 From: Sebastian Borza Date: Fri, 10 Apr 2020 10:50:15 -0500 Subject: [PATCH 18/37] chore(client): ensure UDP port is actually an int. Closes #651. (#815) * chore(client): ensure UDP port is actually an int. Closes #651. * chore(CHANGELOG): update to include PR#651 --- CHANGELOG.md | 1 + influxdb/client.py | 2 +- 2 files changed, 2 insertions(+), 1 deletion(-) diff --git a/CHANGELOG.md b/CHANGELOG.md index 11f47e38..31e36680 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -30,6 +30,7 @@ and this project adheres to [Semantic Versioning](http://semver.org/spec/v2.0.0. - Update DataFrameClient to fix faulty empty tags (#770 thx @michelfripiat) - Update DataFrameClient to properly return `dropna` values (#778 thx @jgspiro) - Update DataFrameClient to test for pd.DataTimeIndex before blind conversion (#623 thx @testforvin) +- Update client to type-set UDP port to int (#651 thx @yifeikong) ### Removed diff --git a/influxdb/client.py b/influxdb/client.py index 390d8e16..0f0350c8 100644 --- a/influxdb/client.py +++ b/influxdb/client.py @@ -107,7 +107,7 @@ def __init__(self, self._verify_ssl = verify_ssl self.__use_udp = use_udp - self.__udp_port = udp_port + self.__udp_port = int(udp_port) if not session: session = requests.Session() From 3e4dc9e31cd900d4a85b8c776b6af91429fc2273 Mon Sep 17 00:00:00 2001 From: Cesar Sanz Date: Fri, 10 Apr 2020 18:28:53 +0200 Subject: [PATCH 19/37] Specify Retention Policy in SeriesHelper (#723) * Allow specify a retention policy in SeriesHelper * Complete the annotation example with the retention policy * Fix formatting * Fix formatting again * Add helper write with retention policy * Add helper write without retention policy * Remove blank line after the docstring. Co-authored-by: Sebastian Borza --- influxdb/helper.py | 8 +++++- influxdb/tests/helper_test.py | 51 +++++++++++++++++++++++++++++++++++ 2 files changed, 58 insertions(+), 1 deletion(-) diff --git a/influxdb/helper.py b/influxdb/helper.py index fa79c079..f49f40ad 100644 --- a/influxdb/helper.py +++ b/influxdb/helper.py @@ -41,6 +41,8 @@ class Meta: # Only applicable if autocommit is True. autocommit = True # If True and no bulk_size, then will set bulk_size to 1. + retention_policy = 'your_retention_policy' + # Specify the retention policy for the data points time_precision = "h"|"m"|s"|"ms"|"u"|"ns" # Default is ns (nanoseconds) # Setting time precision while writing point @@ -83,6 +85,8 @@ def __new__(cls, *args, **kwargs): 'In {0}, time_precision is set, but invalid use any of {}.' .format(cls.__name__, ','.join(allowed_time_precisions))) + cls._retention_policy = getattr(_meta, 'retention_policy', None) + cls._client = getattr(_meta, 'client', None) if cls._autocommit and not cls._client: raise AttributeError( @@ -154,9 +158,11 @@ def commit(cls, client=None): """ if not client: client = cls._client + rtn = client.write_points( cls._json_body_(), - time_precision=cls._time_precision) + time_precision=cls._time_precision, + retention_policy=cls._retention_policy) # will be None if not set and will default to ns cls._reset_() return rtn diff --git a/influxdb/tests/helper_test.py b/influxdb/tests/helper_test.py index 6aa8f15a..16924936 100644 --- a/influxdb/tests/helper_test.py +++ b/influxdb/tests/helper_test.py @@ -376,3 +376,54 @@ class Meta: .format(WarnBulkSizeNoEffect)) self.assertIn('has no affect', str(w[-1].message), 'Warning message did not contain "has not affect".') + + def testSeriesWithRetentionPolicy(self): + """Test that the data is saved with the specified retention policy.""" + my_policy = 'my_policy' + + class RetentionPolicySeriesHelper(SeriesHelper): + + class Meta: + client = InfluxDBClient() + series_name = 'events.stats.{server_name}' + fields = ['some_stat', 'time'] + tags = ['server_name', 'other_tag'] + bulk_size = 2 + autocommit = True + retention_policy = my_policy + + fake_write_points = mock.MagicMock() + RetentionPolicySeriesHelper( + server_name='us.east-1', some_stat=159, other_tag='gg') + RetentionPolicySeriesHelper._client.write_points = fake_write_points + RetentionPolicySeriesHelper( + server_name='us.east-1', some_stat=158, other_tag='aa') + + kall = fake_write_points.call_args + args, kwargs = kall + self.assertTrue('retention_policy' in kwargs) + self.assertEqual(kwargs['retention_policy'], my_policy) + + def testSeriesWithoutRetentionPolicy(self): + """Test that the data is saved without any retention policy.""" + class NoRetentionPolicySeriesHelper(SeriesHelper): + + class Meta: + client = InfluxDBClient() + series_name = 'events.stats.{server_name}' + fields = ['some_stat', 'time'] + tags = ['server_name', 'other_tag'] + bulk_size = 2 + autocommit = True + + fake_write_points = mock.MagicMock() + NoRetentionPolicySeriesHelper( + server_name='us.east-1', some_stat=159, other_tag='gg') + NoRetentionPolicySeriesHelper._client.write_points = fake_write_points + NoRetentionPolicySeriesHelper( + server_name='us.east-1', some_stat=158, other_tag='aa') + + kall = fake_write_points.call_args + args, kwargs = kall + self.assertTrue('retention_policy' in kwargs) + self.assertEqual(kwargs['retention_policy'], None) From 3f2f167c86eedeb27a25e6621145e87a70093c1b Mon Sep 17 00:00:00 2001 From: Sebastian Borza Date: Fri, 10 Apr 2020 11:29:58 -0500 Subject: [PATCH 20/37] chore(CHANGELOG): update for PR#723 --- CHANGELOG.md | 1 + 1 file changed, 1 insertion(+) diff --git a/CHANGELOG.md b/CHANGELOG.md index 31e36680..f2fad368 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -19,6 +19,7 @@ and this project adheres to [Semantic Versioning](http://semver.org/spec/v2.0.0. - Add support for custom request session in InfluxDBClient (#360 thx @dschien) - Add support for handling np.nan and np.inf values in DataFrameClient (#436 thx @nmerket) - Add support for optional `time_precision` in the SeriesHelper (#502 && #719 thx @appunni-dishq && @klDen) +- Add ability to specify retention policy in SeriesHelper (#723 thx @csanz91) ### Changed - Clean up stale CI config (#755) From 61a0fdc3f64532ce1d2ec61264efa556c2a03118 Mon Sep 17 00:00:00 2001 From: Kevin Claytor Date: Fri, 10 Apr 2020 12:49:26 -0400 Subject: [PATCH 21/37] gzip compression for data (post and responses) (#732) * gzip compression working in my influx stack. Needs proper tests. * Also gzip data from server, slightly more straightforward data handling. * Adding in test cases. * Switching back to zlib with gzip headers. * flake8 compatibility * Move parameter into correct position. per review * Switching back to gzip for the headers. * Fixing python 2.7 compatability with gzip. * flake8 compatibility. * flake8 testing Co-authored-by: Kevin Claytor Co-authored-by: Sebastian Borza --- influxdb/client.py | 25 ++++++- influxdb/tests/client_test.py | 67 +++++++++++++++++++ influxdb/tests/server_tests/base.py | 47 +++++++++++++ .../server_tests/client_test_with_server.py | 24 +++++++ 4 files changed, 162 insertions(+), 1 deletion(-) diff --git a/influxdb/client.py b/influxdb/client.py index 0f0350c8..3262c242 100644 --- a/influxdb/client.py +++ b/influxdb/client.py @@ -7,7 +7,9 @@ from __future__ import unicode_literals import datetime +import gzip import itertools +import io import json import random import socket @@ -70,10 +72,11 @@ class InfluxDBClient(object): as a single file containing the private key and the certificate, or as a tuple of both files’ paths, defaults to None :type cert: str + :param gzip: use gzip content encoding to compress requests + :type gzip: bool :param session: allow for the new client request to use an existing requests Session, defaults to None :type session: requests.Session - :raises ValueError: if cert is provided but ssl is disabled (set to False) """ @@ -93,6 +96,7 @@ def __init__(self, pool_size=10, path='', cert=None, + gzip=False, session=None, ): """Construct a new InfluxDBClient object.""" @@ -159,6 +163,8 @@ def __init__(self, 'Accept': 'application/x-msgpack' } + self._gzip = gzip + @property def _baseurl(self): return self.__baseurl @@ -278,6 +284,23 @@ def request(self, url, method='GET', params=None, data=None, if isinstance(data, (dict, list)): data = json.dumps(data) + if self._gzip: + # Receive and send compressed data + headers.update({ + 'Accept-Encoding': 'gzip', + 'Content-Encoding': 'gzip', + }) + if data is not None: + # For Py 2.7 compatability use Gzipfile + compressed = io.BytesIO() + with gzip.GzipFile( + compresslevel=9, + fileobj=compressed, + mode='w' + ) as f: + f.write(data) + data = compressed.getvalue() + # Try to send the request more than once by default (see #103) retry = True _try = 0 diff --git a/influxdb/tests/client_test.py b/influxdb/tests/client_test.py index fe7381ea..f8f3cb00 100644 --- a/influxdb/tests/client_test.py +++ b/influxdb/tests/client_test.py @@ -24,6 +24,8 @@ import unittest import warnings +import io +import gzip import json import mock import requests @@ -214,6 +216,71 @@ def test_write_points(self): m.last_request.body.decode('utf-8'), ) + def test_write_gzip(self): + """Test write in TestInfluxDBClient object.""" + with requests_mock.Mocker() as m: + m.register_uri( + requests_mock.POST, + "http://localhost:8086/write", + status_code=204 + ) + + cli = InfluxDBClient(database='db', gzip=True) + cli.write( + {"database": "mydb", + "retentionPolicy": "mypolicy", + "points": [{"measurement": "cpu_load_short", + "tags": {"host": "server01", + "region": "us-west"}, + "time": "2009-11-10T23:00:00Z", + "fields": {"value": 0.64}}]} + ) + + compressed = io.BytesIO() + with gzip.GzipFile( + compresslevel=9, + fileobj=compressed, + mode='w' + ) as f: + f.write( + b"cpu_load_short,host=server01,region=us-west " + b"value=0.64 1257894000000000000\n" + ) + + self.assertEqual( + m.last_request.body, + compressed.getvalue(), + ) + + def test_write_points_gzip(self): + """Test write points for TestInfluxDBClient object.""" + with requests_mock.Mocker() as m: + m.register_uri( + requests_mock.POST, + "http://localhost:8086/write", + status_code=204 + ) + + cli = InfluxDBClient(database='db', gzip=True) + cli.write_points( + self.dummy_points, + ) + + compressed = io.BytesIO() + with gzip.GzipFile( + compresslevel=9, + fileobj=compressed, + mode='w' + ) as f: + f.write( + b'cpu_load_short,host=server01,region=us-west ' + b'value=0.64 1257894000123456000\n' + ) + self.assertEqual( + m.last_request.body, + compressed.getvalue(), + ) + def test_write_points_toplevel_attributes(self): """Test write points attrs for TestInfluxDBClient object.""" with requests_mock.Mocker() as m: diff --git a/influxdb/tests/server_tests/base.py b/influxdb/tests/server_tests/base.py index fe722870..45a9ec80 100644 --- a/influxdb/tests/server_tests/base.py +++ b/influxdb/tests/server_tests/base.py @@ -36,6 +36,15 @@ def _setup_influxdb_server(inst): database='db') +def _setup_gzip_client(inst): + inst.cli = InfluxDBClient('localhost', + inst.influxd_inst.http_port, + 'root', + '', + database='db', + gzip=True) + + def _teardown_influxdb_server(inst): remove_tree = sys.exc_info() == (None, None, None) inst.influxd_inst.close(remove_tree=remove_tree) @@ -89,3 +98,41 @@ def tearDownClass(cls): def tearDown(self): """Deconstruct an instance of ManyTestCasesWithServerMixin.""" self.cli.drop_database('db') + + +class SingleTestCaseWithServerGzipMixin(object): + """Define the single testcase with server with gzip client mixin. + + Same as the SingleTestCaseWithServerGzipMixin but the InfluxDBClient has + gzip=True + """ + + @classmethod + def setUp(cls): + """Set up an instance of the SingleTestCaseWithServerGzipMixin.""" + _setup_influxdb_server(cls) + _setup_gzip_client(cls) + + @classmethod + def tearDown(cls): + """Tear down an instance of the SingleTestCaseWithServerMixin.""" + _teardown_influxdb_server(cls) + + +class ManyTestCasesWithServerGzipMixin(object): + """Define the many testcase with server with gzip client mixin. + + Same as the ManyTestCasesWithServerMixin but the InfluxDBClient has + gzip=True. + """ + + @classmethod + def setUpClass(cls): + """Set up an instance of the ManyTestCasesWithServerGzipMixin.""" + _setup_influxdb_server(cls) + _setup_gzip_client(cls) + + @classmethod + def tearDown(cls): + """Tear down an instance of the SingleTestCaseWithServerMixin.""" + _teardown_influxdb_server(cls) diff --git a/influxdb/tests/server_tests/client_test_with_server.py b/influxdb/tests/server_tests/client_test_with_server.py index 94f28b66..020014c3 100644 --- a/influxdb/tests/server_tests/client_test_with_server.py +++ b/influxdb/tests/server_tests/client_test_with_server.py @@ -26,6 +26,8 @@ from influxdb.tests import skip_if_pypy, using_pypy, skip_server_tests from influxdb.tests.server_tests.base import ManyTestCasesWithServerMixin from influxdb.tests.server_tests.base import SingleTestCaseWithServerMixin +from influxdb.tests.server_tests.base import ManyTestCasesWithServerGzipMixin +from influxdb.tests.server_tests.base import SingleTestCaseWithServerGzipMixin # By default, raise exceptions on warnings warnings.simplefilter('error', FutureWarning) @@ -913,3 +915,25 @@ def test_write_points_udp(self): ], list(rsp['cpu_load_short']) ) + + +# Run the tests again, but with gzip enabled this time +@skip_server_tests +class GzipSimpleTests(SimpleTests, SingleTestCaseWithServerGzipMixin): + """Repeat the simple tests with InfluxDBClient where gzip=True.""" + + pass + + +@skip_server_tests +class GzipCommonTests(CommonTests, ManyTestCasesWithServerGzipMixin): + """Repeat the common tests with InfluxDBClient where gzip=True.""" + + pass + + +@skip_server_tests +class GzipUdpTests(UdpTests, ManyTestCasesWithServerGzipMixin): + """Repeat the UDP tests with InfluxDBClient where gzip=True.""" + + pass From 9970ed0cce527ee1519def3cb5ec2c323855c476 Mon Sep 17 00:00:00 2001 From: Sebastian Borza Date: Fri, 10 Apr 2020 11:50:38 -0500 Subject: [PATCH 22/37] chore(CHANGELOG): update to add PR#732 --- CHANGELOG.md | 1 + 1 file changed, 1 insertion(+) diff --git a/CHANGELOG.md b/CHANGELOG.md index f2fad368..3a706e7b 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -20,6 +20,7 @@ and this project adheres to [Semantic Versioning](http://semver.org/spec/v2.0.0. - Add support for handling np.nan and np.inf values in DataFrameClient (#436 thx @nmerket) - Add support for optional `time_precision` in the SeriesHelper (#502 && #719 thx @appunni-dishq && @klDen) - Add ability to specify retention policy in SeriesHelper (#723 thx @csanz91) +- Add gzip compression for post and response data (#732 thx @KEClaytor) ### Changed - Clean up stale CI config (#755) From ea5b1d56060cd87e7039fc30c630d606cc129ede Mon Sep 17 00:00:00 2001 From: Jamie Hewland Date: Fri, 10 Apr 2020 18:58:31 +0200 Subject: [PATCH 23/37] Make batched writing support all iterables (#746) * Make batched writing support all iterables * Also test batching generator against real server * Fix PEP257 error * Import itertools functions directly --- influxdb/client.py | 15 ++++++++-- influxdb/tests/client_test.py | 30 +++++++++++++++++++ .../server_tests/client_test_with_server.py | 27 +++++++++++++++++ 3 files changed, 69 insertions(+), 3 deletions(-) diff --git a/influxdb/client.py b/influxdb/client.py index 3262c242..46424bc2 100644 --- a/influxdb/client.py +++ b/influxdb/client.py @@ -15,11 +15,11 @@ import socket import struct import time +from itertools import chain, islice import msgpack import requests import requests.exceptions -from six.moves import xrange from six.moves.urllib.parse import urlparse from influxdb.line_protocol import make_lines, quote_ident, quote_literal @@ -597,8 +597,17 @@ def ping(self): @staticmethod def _batches(iterable, size): - for i in xrange(0, len(iterable), size): - yield iterable[i:i + size] + # Iterate over an iterable producing iterables of batches. Based on: + # http://code.activestate.com/recipes/303279-getting-items-in-batches/ + iterator = iter(iterable) + while True: + try: # Try get the first element in the iterator... + head = (next(iterator),) + except StopIteration: + return # ...so that we can stop if there isn't one + # Otherwise, lazily slice the rest of the batch + rest = islice(iterator, size - 1) + yield chain(head, rest) def _write_points(self, points, diff --git a/influxdb/tests/client_test.py b/influxdb/tests/client_test.py index f8f3cb00..a8f8e864 100644 --- a/influxdb/tests/client_test.py +++ b/influxdb/tests/client_test.py @@ -332,6 +332,36 @@ def test_write_points_batch(self): self.assertEqual(expected_last_body, m.last_request.body.decode('utf-8')) + def test_write_points_batch_generator(self): + """Test write points batch from a generator for TestInfluxDBClient.""" + dummy_points = [ + {"measurement": "cpu_usage", "tags": {"unit": "percent"}, + "time": "2009-11-10T23:00:00Z", "fields": {"value": 12.34}}, + {"measurement": "network", "tags": {"direction": "in"}, + "time": "2009-11-10T23:00:00Z", "fields": {"value": 123.00}}, + {"measurement": "network", "tags": {"direction": "out"}, + "time": "2009-11-10T23:00:00Z", "fields": {"value": 12.00}} + ] + dummy_points_generator = (point for point in dummy_points) + expected_last_body = ( + "network,direction=out,host=server01,region=us-west " + "value=12.0 1257894000000000000\n" + ) + + with requests_mock.Mocker() as m: + m.register_uri(requests_mock.POST, + "http://localhost:8086/write", + status_code=204) + cli = InfluxDBClient(database='db') + cli.write_points(points=dummy_points_generator, + database='db', + tags={"host": "server01", + "region": "us-west"}, + batch_size=2) + self.assertEqual(m.call_count, 2) + self.assertEqual(expected_last_body, + m.last_request.body.decode('utf-8')) + def test_write_points_udp(self): """Test write points UDP for TestInfluxDBClient object.""" s = socket.socket(socket.AF_INET, socket.SOCK_DGRAM) diff --git a/influxdb/tests/server_tests/client_test_with_server.py b/influxdb/tests/server_tests/client_test_with_server.py index 020014c3..a0263243 100644 --- a/influxdb/tests/server_tests/client_test_with_server.py +++ b/influxdb/tests/server_tests/client_test_with_server.py @@ -452,6 +452,33 @@ def test_write_points_batch(self): self.assertIn(12, net_out['series'][0]['values'][0]) self.assertIn(12.34, cpu['series'][0]['values'][0]) + def test_write_points_batch_generator(self): + """Test writing points in a batch from a generator.""" + dummy_points = [ + {"measurement": "cpu_usage", "tags": {"unit": "percent"}, + "time": "2009-11-10T23:00:00Z", "fields": {"value": 12.34}}, + {"measurement": "network", "tags": {"direction": "in"}, + "time": "2009-11-10T23:00:00Z", "fields": {"value": 123.00}}, + {"measurement": "network", "tags": {"direction": "out"}, + "time": "2009-11-10T23:00:00Z", "fields": {"value": 12.00}} + ] + dummy_points_generator = (point for point in dummy_points) + self.cli.write_points(points=dummy_points_generator, + tags={"host": "server01", + "region": "us-west"}, + batch_size=2) + time.sleep(5) + net_in = self.cli.query("SELECT value FROM network " + "WHERE direction=$dir", + bind_params={'dir': 'in'} + ).raw + net_out = self.cli.query("SELECT value FROM network " + "WHERE direction='out'").raw + cpu = self.cli.query("SELECT value FROM cpu_usage").raw + self.assertIn(123, net_in['series'][0]['values'][0]) + self.assertIn(12, net_out['series'][0]['values'][0]) + self.assertIn(12.34, cpu['series'][0]['values'][0]) + def test_query(self): """Test querying data back from server.""" self.assertIs(True, self.cli.write_points(dummy_point)) From 72dabb7ba6e3ae12061084166e36bb436232d1af Mon Sep 17 00:00:00 2001 From: Sebastian Borza Date: Fri, 10 Apr 2020 11:59:25 -0500 Subject: [PATCH 24/37] chore(CHANGELOG): update to include PR#746 --- CHANGELOG.md | 1 + 1 file changed, 1 insertion(+) diff --git a/CHANGELOG.md b/CHANGELOG.md index 3a706e7b..5a49c84a 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -33,6 +33,7 @@ and this project adheres to [Semantic Versioning](http://semver.org/spec/v2.0.0. - Update DataFrameClient to properly return `dropna` values (#778 thx @jgspiro) - Update DataFrameClient to test for pd.DataTimeIndex before blind conversion (#623 thx @testforvin) - Update client to type-set UDP port to int (#651 thx @yifeikong) +- Update batched writing support for all iterables (#746 thx @JayH5) ### Removed From 8f6bfaa2ddd67aa1faad202fc6ac0d92d38d26b4 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Stefan=20=22hr=22=20Berder=20=28=E7=99=BD=E5=B3=B0=29?= Date: Sat, 11 Apr 2020 01:10:40 +0800 Subject: [PATCH 25/37] Fix chunked query to return chunk resultsets (#753) When querying large data sets, it's vital to get a chunked responses to manage memory usage. Wrapping the query response in a generator and streaming the request provides the desired result. It also fixes `InfluxDBClient.query()` behavior for chunked queries that is currently not working according to [specs](https://github.com/influxdata/influxdb-python/blob/master/influxdb/client.py#L429) Closes #585. Closes #531. Closes #538. --- influxdb/client.py | 10 +++++--- influxdb/tests/client_test.py | 44 ++++++++++++++--------------------- 2 files changed, 25 insertions(+), 29 deletions(-) diff --git a/influxdb/client.py b/influxdb/client.py index 46424bc2..b28ed1b5 100644 --- a/influxdb/client.py +++ b/influxdb/client.py @@ -249,7 +249,7 @@ def switch_user(self, username, password): self._username = username self._password = password - def request(self, url, method='GET', params=None, data=None, + def request(self, url, method='GET', params=None, data=None, stream=False, expected_response_code=200, headers=None): """Make a HTTP request to the InfluxDB API. @@ -261,6 +261,8 @@ def request(self, url, method='GET', params=None, data=None, :type params: dict :param data: the data of the request, defaults to None :type data: str + :param stream: True if a query uses chunked responses + :type stream: bool :param expected_response_code: the expected response code of the request, defaults to 200 :type expected_response_code: int @@ -312,6 +314,7 @@ def request(self, url, method='GET', params=None, data=None, auth=(self._username, self._password), params=params, data=data, + stream=stream, headers=headers, proxies=self._proxies, verify=self._verify_ssl, @@ -398,17 +401,17 @@ def write(self, data, params=None, expected_response_code=204, @staticmethod def _read_chunked_response(response, raise_errors=True): - result_set = {} for line in response.iter_lines(): if isinstance(line, bytes): line = line.decode('utf-8') data = json.loads(line) + result_set = {} for result in data.get('results', []): for _key in result: if isinstance(result[_key], list): result_set.setdefault( _key, []).extend(result[_key]) - return ResultSet(result_set, raise_errors=raise_errors) + yield ResultSet(result_set, raise_errors=raise_errors) def query(self, query, @@ -499,6 +502,7 @@ def query(self, method=method, params=params, data=None, + stream=chunked, expected_response_code=expected_response_code ) diff --git a/influxdb/tests/client_test.py b/influxdb/tests/client_test.py index a8f8e864..fd3c06bb 100644 --- a/influxdb/tests/client_test.py +++ b/influxdb/tests/client_test.py @@ -1400,16 +1400,11 @@ def test_invalid_port_fails(self): def test_chunked_response(self): """Test chunked reponse for TestInfluxDBClient object.""" example_response = \ - u'{"results":[{"statement_id":0,"series":' \ - '[{"name":"cpu","columns":["fieldKey","fieldType"],"values":' \ - '[["value","integer"]]}],"partial":true}]}\n{"results":' \ - '[{"statement_id":0,"series":[{"name":"iops","columns":' \ - '["fieldKey","fieldType"],"values":[["value","integer"]]}],' \ - '"partial":true}]}\n{"results":[{"statement_id":0,"series":' \ - '[{"name":"load","columns":["fieldKey","fieldType"],"values":' \ - '[["value","integer"]]}],"partial":true}]}\n{"results":' \ - '[{"statement_id":0,"series":[{"name":"memory","columns":' \ - '["fieldKey","fieldType"],"values":[["value","integer"]]}]}]}\n' + u'{"results":[{"statement_id":0,"series":[{"columns":["key"],' \ + '"values":[["cpu"],["memory"],["iops"],["network"]],"partial":' \ + 'true}],"partial":true}]}\n{"results":[{"statement_id":0,' \ + '"series":[{"columns":["key"],"values":[["qps"],["uptime"],' \ + '["df"],["mount"]]}]}]}\n' with requests_mock.Mocker() as m: m.register_uri( @@ -1417,23 +1412,20 @@ def test_chunked_response(self): "http://localhost:8086/query", text=example_response ) - response = self.cli.query('show series limit 4 offset 0', + response = self.cli.query('show series', chunked=True, chunk_size=4) - self.assertTrue(len(response) == 4) - self.assertEqual(response.__repr__(), ResultSet( - {'series': [{'values': [['value', 'integer']], - 'name': 'cpu', - 'columns': ['fieldKey', 'fieldType']}, - {'values': [['value', 'integer']], - 'name': 'iops', - 'columns': ['fieldKey', 'fieldType']}, - {'values': [['value', 'integer']], - 'name': 'load', - 'columns': ['fieldKey', 'fieldType']}, - {'values': [['value', 'integer']], - 'name': 'memory', - 'columns': ['fieldKey', 'fieldType']}]} - ).__repr__()) + res = list(response) + self.assertTrue(len(res) == 2) + self.assertEqual(res[0].__repr__(), ResultSet( + {'series': [{ + 'columns': ['key'], + 'values': [['cpu'], ['memory'], ['iops'], ['network']] + }]}).__repr__()) + self.assertEqual(res[1].__repr__(), ResultSet( + {'series': [{ + 'columns': ['key'], + 'values': [['qps'], ['uptime'], ['df'], ['mount']] + }]}).__repr__()) class FakeClient(InfluxDBClient): From 2082f23a2d775f888dc5f731c79db1007328c663 Mon Sep 17 00:00:00 2001 From: Sebastian Borza Date: Fri, 10 Apr 2020 12:12:03 -0500 Subject: [PATCH 26/37] chore(CHANGELOG): update to include PR#753 and #538 --- CHANGELOG.md | 1 + 1 file changed, 1 insertion(+) diff --git a/CHANGELOG.md b/CHANGELOG.md index 5a49c84a..00d2a4f7 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -21,6 +21,7 @@ and this project adheres to [Semantic Versioning](http://semver.org/spec/v2.0.0. - Add support for optional `time_precision` in the SeriesHelper (#502 && #719 thx @appunni-dishq && @klDen) - Add ability to specify retention policy in SeriesHelper (#723 thx @csanz91) - Add gzip compression for post and response data (#732 thx @KEClaytor) +- Add support for chunked responses in ResultSet (#753 and #538 thx @hrbonz && @psy0rz) ### Changed - Clean up stale CI config (#755) From 0b8b6a1647b22e460a24db67778a057f917acbc9 Mon Sep 17 00:00:00 2001 From: Greg Schrock Date: Fri, 10 Apr 2020 13:29:54 -0400 Subject: [PATCH 27/37] Fix make_lines excludes fields with empty strings (#655) (#766) * Fix make_lines excludes fields with empty strings (#655) Converting to unicode required something to be done with None values. They were converted to empty strings which were subsequently ignored. This makes it impossible to write an explicitly empty string, which should be possible. This change distinguishes between None and empty strings. * Fix linting failure due to long comment line Co-authored-by: Greg Schrock --- influxdb/line_protocol.py | 6 ++++-- influxdb/tests/test_line_protocol.py | 18 ++++++++++++++++++ 2 files changed, 22 insertions(+), 2 deletions(-) diff --git a/influxdb/line_protocol.py b/influxdb/line_protocol.py index d6cbf46f..25dd2ad7 100644 --- a/influxdb/line_protocol.py +++ b/influxdb/line_protocol.py @@ -104,9 +104,11 @@ def _is_float(value): def _escape_value(value): - value = _get_unicode(value) + if value is None: + return '' - if isinstance(value, text_type) and value != '': + value = _get_unicode(value) + if isinstance(value, text_type): return quote_ident(value) if isinstance(value, integer_types) and not isinstance(value, bool): diff --git a/influxdb/tests/test_line_protocol.py b/influxdb/tests/test_line_protocol.py index c48d5edc..5b344990 100644 --- a/influxdb/tests/test_line_protocol.py +++ b/influxdb/tests/test_line_protocol.py @@ -117,6 +117,24 @@ def test_make_lines_unicode(self): 'test,unicode_tag=\'Привет!\' unicode_val="Привет!"\n' ) + def test_make_lines_empty_field_string(self): + """Test make lines with an empty string field.""" + data = { + "points": [ + { + "measurement": "test", + "fields": { + "string": "", + } + } + ] + } + + self.assertEqual( + line_protocol.make_lines(data), + 'test string=""\n' + ) + def test_tag_value_newline(self): """Test make lines with tag value contains newline.""" data = { From 13aa3a36bee854e89a0993d8e9d3f142fd504a25 Mon Sep 17 00:00:00 2001 From: Sebastian Borza Date: Fri, 10 Apr 2020 12:30:50 -0500 Subject: [PATCH 28/37] chore(CHANGELOG): update for PR #766 --- CHANGELOG.md | 1 + 1 file changed, 1 insertion(+) diff --git a/CHANGELOG.md b/CHANGELOG.md index 00d2a4f7..4ee39303 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -22,6 +22,7 @@ and this project adheres to [Semantic Versioning](http://semver.org/spec/v2.0.0. - Add ability to specify retention policy in SeriesHelper (#723 thx @csanz91) - Add gzip compression for post and response data (#732 thx @KEClaytor) - Add support for chunked responses in ResultSet (#753 and #538 thx @hrbonz && @psy0rz) +- Add support for empty string fields (#766 thx @gregschrock) ### Changed - Clean up stale CI config (#755) From dfdfe267b2ee74bd65b6bbd4afdd52a979ba4bd2 Mon Sep 17 00:00:00 2001 From: Keunhyun Oh Date: Sat, 11 Apr 2020 02:33:14 +0900 Subject: [PATCH 29/37] fix: Calling commit and _json_body make raising an exception when any datapoints are not added. (#772) * If any SerialHelper is not generated, calling the commit function makes raising an exception because _datapoints is not allocated. It is hard to find the reason of this error. it is because reviewing influxdb-python's code is needed. I think that it is important that is producing predictable results. Results when calling first time is needed to equal to calling resetting datapoints in json_body. So, I've fixed that if not initialized when calling _json_body() function, _datapoints is reset to avoid raising error. In Unittest, the setup function is added. When calling setup function firstly, __initialized__ is False and _datapoints is not assigned. But, because of this commit, it is OK. Contacts: Keunhyun Oh * fix build fail Contacts: Keunhyun Oh * fix build fail Contacts: Keunhyun Oh * fix build fail Contacts: Keunhyun Oh * Update helper_test.py * Update helper_test.py * Update helper_test.py --- influxdb/helper.py | 2 ++ influxdb/influxdb08/helper.py | 2 ++ influxdb/tests/helper_test.py | 8 ++++++++ 3 files changed, 12 insertions(+) diff --git a/influxdb/helper.py b/influxdb/helper.py index f49f40ad..74209354 100644 --- a/influxdb/helper.py +++ b/influxdb/helper.py @@ -174,6 +174,8 @@ def _json_body_(cls): :return: JSON body of these datapoints. """ json = [] + if not cls.__initialized__: + cls._reset_() for series_name, data in six.iteritems(cls._datapoints): for point in data: json_point = { diff --git a/influxdb/influxdb08/helper.py b/influxdb/influxdb08/helper.py index f3dec33c..5f2d4614 100644 --- a/influxdb/influxdb08/helper.py +++ b/influxdb/influxdb08/helper.py @@ -139,6 +139,8 @@ def _json_body_(cls): :return: JSON body of the datapoints. """ json = [] + if not cls.__initialized__: + cls._reset_() for series_name, data in six.iteritems(cls._datapoints): json.append({'name': series_name, 'columns': cls._fields, diff --git a/influxdb/tests/helper_test.py b/influxdb/tests/helper_test.py index 16924936..6737f921 100644 --- a/influxdb/tests/helper_test.py +++ b/influxdb/tests/helper_test.py @@ -47,6 +47,14 @@ class Meta: TestSeriesHelper.MySeriesHelper = MySeriesHelper + def setUp(self): + """Check that MySeriesHelper has empty datapoints.""" + super(TestSeriesHelper, self).setUp() + self.assertEqual( + TestSeriesHelper.MySeriesHelper._json_body_(), + [], + 'Resetting helper in teardown did not empty datapoints.') + def tearDown(self): """Deconstruct the TestSeriesHelper object.""" super(TestSeriesHelper, self).tearDown() From 21379a503922197f93c902df9ce9ed5260db0042 Mon Sep 17 00:00:00 2001 From: Sebastian Borza Date: Fri, 10 Apr 2020 12:34:05 -0500 Subject: [PATCH 30/37] chore(CHANGELOG): add PR#772 --- CHANGELOG.md | 1 + 1 file changed, 1 insertion(+) diff --git a/CHANGELOG.md b/CHANGELOG.md index 4ee39303..322c952b 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -36,6 +36,7 @@ and this project adheres to [Semantic Versioning](http://semver.org/spec/v2.0.0. - Update DataFrameClient to test for pd.DataTimeIndex before blind conversion (#623 thx @testforvin) - Update client to type-set UDP port to int (#651 thx @yifeikong) - Update batched writing support for all iterables (#746 thx @JayH5) +- Update SeriesHelper to enable class instantiation when not initialized (#772 thx @ocworld) ### Removed From 23f1d32fcfe1a4ce316ae8cfd3511205a6951ef8 Mon Sep 17 00:00:00 2001 From: Shan Desai Date: Fri, 10 Apr 2020 19:59:04 +0200 Subject: [PATCH 31/37] Refactor `tutorial_udp` example for timestamps (#808) - `time` key should be within each datapoint - This PR addresses #788 regarding the structure of the data when inserting via UDP. - The original documentation contributed by me took the structure of the `tutorial.py` as base. However, upon testing, the timestamp in the example are not written (2020 is written as opposed to 2009). - Tested for `influxdb-python` v5.2.3 and InfluxDB v1.6.1 --- examples/tutorial_udp.py | 15 ++++++++------- 1 file changed, 8 insertions(+), 7 deletions(-) diff --git a/examples/tutorial_udp.py b/examples/tutorial_udp.py index 517ae858..93b923d7 100644 --- a/examples/tutorial_udp.py +++ b/examples/tutorial_udp.py @@ -29,18 +29,19 @@ def main(uport): "host": "server01", "region": "us-west" }, - "time": "2009-11-10T23:00:00Z", "points": [{ "measurement": "cpu_load_short", "fields": { "value": 0.64 - } + }, + "time": "2009-11-10T23:00:00Z", }, - { - "measurement": "cpu_load_short", - "fields": { - "value": 0.67 - } + { + "measurement": "cpu_load_short", + "fields": { + "value": 0.67 + }, + "time": "2009-11-10T23:05:00Z" }] } From a6ec2545842a61f7b3cfe4c3a5d8fbe4109110fb Mon Sep 17 00:00:00 2001 From: Sebastian Borza Date: Fri, 10 Apr 2020 14:27:14 -0500 Subject: [PATCH 32/37] feat(client): add support for context managers (#816) * feat(client): add support for context managers * chore(CHANGELOG): rebase against master --- CHANGELOG.md | 2 ++ influxdb/client.py | 12 ++++++++++++ 2 files changed, 14 insertions(+) diff --git a/CHANGELOG.md b/CHANGELOG.md index 322c952b..411b46a4 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -23,6 +23,7 @@ and this project adheres to [Semantic Versioning](http://semver.org/spec/v2.0.0. - Add gzip compression for post and response data (#732 thx @KEClaytor) - Add support for chunked responses in ResultSet (#753 and #538 thx @hrbonz && @psy0rz) - Add support for empty string fields (#766 thx @gregschrock) +- Add support for context managers to InfluxDBClient (#721 thx @JustusAdam) ### Changed - Clean up stale CI config (#755) @@ -37,6 +38,7 @@ and this project adheres to [Semantic Versioning](http://semver.org/spec/v2.0.0. - Update client to type-set UDP port to int (#651 thx @yifeikong) - Update batched writing support for all iterables (#746 thx @JayH5) - Update SeriesHelper to enable class instantiation when not initialized (#772 thx @ocworld) +- Update UDP test case to add proper timestamp to datapoints (#808 thx @shantanoo-desai) ### Removed diff --git a/influxdb/client.py b/influxdb/client.py index b28ed1b5..a0f571f5 100644 --- a/influxdb/client.py +++ b/influxdb/client.py @@ -35,6 +35,9 @@ class InfluxDBClient(object): connect to InfluxDB. Requests can be made to InfluxDB directly through the client. + The client supports the use as a `context manager + `_. + :param host: hostname to connect to InfluxDB, defaults to 'localhost' :type host: str :param port: port to connect to InfluxDB, defaults to 8086 @@ -78,6 +81,7 @@ class InfluxDBClient(object): requests Session, defaults to None :type session: requests.Session :raises ValueError: if cert is provided but ssl is disabled (set to False) + """ def __init__(self, @@ -165,6 +169,14 @@ def __init__(self, self._gzip = gzip + def __enter__(self): + """Enter function as used by context manager.""" + pass + + def __exit__(self, _exc_type, _exc_value, _traceback): + """Exit function as used by context manager.""" + self.close() + @property def _baseurl(self): return self.__baseurl From e2f8579720f9804da53bdb75bf378e6659441ec7 Mon Sep 17 00:00:00 2001 From: Sebastian Borza Date: Fri, 10 Apr 2020 14:38:30 -0500 Subject: [PATCH 33/37] chore(CHANGELOG): tagging release to 5.3.0 --- LICENSE | 2 +- influxdb/__init__.py | 2 +- 2 files changed, 2 insertions(+), 2 deletions(-) diff --git a/LICENSE b/LICENSE index 38ee2491..a49a5410 100644 --- a/LICENSE +++ b/LICENSE @@ -1,6 +1,6 @@ The MIT License (MIT) -Copyright (c) 2013 InfluxDB +Copyright (c) 2020 InfluxDB Permission is hereby granted, free of charge, to any person obtaining a copy of this software and associated documentation files (the "Software"), to deal in diff --git a/influxdb/__init__.py b/influxdb/__init__.py index b31170bb..56f2f619 100644 --- a/influxdb/__init__.py +++ b/influxdb/__init__.py @@ -18,4 +18,4 @@ ] -__version__ = '5.2.3' +__version__ = '5.3.0' From 7105bcde1d9be7f93135f5bb62c2d1e39a01f9d1 Mon Sep 17 00:00:00 2001 From: Krzysztof Baranski Date: Sun, 12 Apr 2020 01:45:32 +0200 Subject: [PATCH 34/37] do not sleep after last retry before raising exception (#790) * do not sleep after last retry before raising exception * documentation: clarification of retry parameter retry=0 - retry forever retry=1 - try once, on error don't do any retry retry=2 - 2 tries, one original and one retry on error retry=3 - 3 tries, one original and maximum two retries on errors * retries - move raise before sleep * retries - documentation * fix line length --- influxdb/client.py | 12 ++++++++---- 1 file changed, 8 insertions(+), 4 deletions(-) diff --git a/influxdb/client.py b/influxdb/client.py index a0f571f5..80994190 100644 --- a/influxdb/client.py +++ b/influxdb/client.py @@ -59,8 +59,12 @@ class InfluxDBClient(object): :param timeout: number of seconds Requests will wait for your client to establish a connection, defaults to None :type timeout: int - :param retries: number of retries your client will try before aborting, - defaults to 3. 0 indicates try until success + :param retries: number of attempts your client will make before aborting, + defaults to 3 + 0 - try until success + 1 - attempt only once (without retry) + 2 - maximum two attempts (including one retry) + 3 - maximum three attempts (default option) :type retries: int :param use_udp: use UDP to connect to InfluxDB, defaults to False :type use_udp: bool @@ -339,10 +343,10 @@ def request(self, url, method='GET', params=None, data=None, stream=False, _try += 1 if self._retries != 0: retry = _try < self._retries - if method == "POST": - time.sleep((2 ** _try) * random.random() / 100.0) if not retry: raise + if method == "POST": + time.sleep((2 ** _try) * random.random() / 100.0) type_header = response.headers and response.headers.get("Content-Type") if type_header == "application/x-msgpack" and response.content: From dd89c2ab23424b2f6edd182bd7a4b794de72a780 Mon Sep 17 00:00:00 2001 From: Sebastian Borza Date: Sat, 11 Apr 2020 18:48:26 -0500 Subject: [PATCH 35/37] chore(CHANGELOG): add PR#790 --- CHANGELOG.md | 5 ++++- 1 file changed, 4 insertions(+), 1 deletion(-) diff --git a/CHANGELOG.md b/CHANGELOG.md index 411b46a4..c2f3edcc 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -4,11 +4,14 @@ All notable changes to this project will be documented in this file. The format is based on [Keep a Changelog](http://keepachangelog.com/en/1.0.0/) and this project adheres to [Semantic Versioning](http://semver.org/spec/v2.0.0.html). -## [Unreleased] +## [v5.3.1] - Unreleased ### Added ### Changed +- Amend retry to avoid sleep after last retry before raising exception (#790 thx @krzysbaranski) + +### Removed ## [v5.3.0] - 2020-04-10 From ed1d233b9c4401e7ae365544db190b97560ad353 Mon Sep 17 00:00:00 2001 From: Matthew Thode Date: Sat, 11 Apr 2020 18:49:33 -0500 Subject: [PATCH 36/37] remove msgpack pin (#818) The hard lock prevents this from being co-installed with many other packages. For instance, it's preventing it from being included in openstack (which is on 0.6.2 and working on 1.0.0 now). Signed-off-by: Matthew Thode --- requirements.txt | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/requirements.txt b/requirements.txt index 77d7306f..548b17c8 100644 --- a/requirements.txt +++ b/requirements.txt @@ -2,4 +2,4 @@ python-dateutil>=2.6.0 pytz requests>=2.17.0 six>=1.10.0 -msgpack==0.6.1 +msgpack From 6a36102c3b78bd3fe303680ede7cfaa8fe979541 Mon Sep 17 00:00:00 2001 From: Sebastian Borza Date: Sat, 11 Apr 2020 18:53:49 -0500 Subject: [PATCH 37/37] chore(CHANGELOG): add PR#818 --- CHANGELOG.md | 1 + 1 file changed, 1 insertion(+) diff --git a/CHANGELOG.md b/CHANGELOG.md index c2f3edcc..d380b73e 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -10,6 +10,7 @@ and this project adheres to [Semantic Versioning](http://semver.org/spec/v2.0.0. ### Changed - Amend retry to avoid sleep after last retry before raising exception (#790 thx @krzysbaranski) +- Remove msgpack pinning for requirements (#818 thx @prometheanfire) ### Removed