diff --git a/CHANGELOG.md b/CHANGELOG.md index f3e86086..bfd27d38 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -4,7 +4,12 @@ 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). -## [v5.3.1] - Unreleased +## [v5.3.2] - 2024-04-17 + +### Changed +- Correctly serialize nanosecond dataframe timestamps (#926) + +## [v5.3.1] - 2022-11-14 ### Added - Add support for custom headers in the InfluxDBClient (#710 thx @nathanielatom) diff --git a/README.rst b/README.rst index a40ed148..b78e626d 100644 --- a/README.rst +++ b/README.rst @@ -1,3 +1,14 @@ +The v1 client libraries for InfluxDB were typically developed and maintained by +community members. For InfluxDB 3.0 users, this library is succeeded by the +lightweight `v3 client library `_. +For InfluxDB 2.0 users, look at the `v2 client library +`_. + +If there are still users of this v1 client library, and they or somebody else +are willing to keep them updated with security fixes at a minimum please reach +out on the `Community Forums `_ or +`InfluxData Slack `_. + InfluxDB-Python =============== @@ -17,6 +28,8 @@ InfluxDB-Python InfluxDB-Python is a client for interacting with InfluxDB_. +**Note: This library is for use with InfluxDB 1.x. For connecting to InfluxDB 2.x instances, please use the the** `influxdb-client-python `_ **client.** + Development of this library is maintained by: +-----------+-------------------------------+ diff --git a/influxdb/__init__.py b/influxdb/__init__.py index 59916c26..e66f80ea 100644 --- a/influxdb/__init__.py +++ b/influxdb/__init__.py @@ -18,4 +18,4 @@ ] -__version__ = '5.3.1' +__version__ = '5.3.2' diff --git a/influxdb/_dataframe_client.py b/influxdb/_dataframe_client.py index e7ae9c17..907db2cb 100644 --- a/influxdb/_dataframe_client.py +++ b/influxdb/_dataframe_client.py @@ -372,10 +372,10 @@ def _convert_dataframe_to_lines(self, # Make array of timestamp ints if isinstance(dataframe.index, pd.PeriodIndex): - time = ((dataframe.index.to_timestamp().values.astype(np.int64) / + time = ((dataframe.index.to_timestamp().values.astype(np.int64) // precision_factor).astype(np.int64).astype(str)) else: - time = ((pd.to_datetime(dataframe.index).values.astype(np.int64) / + time = ((pd.to_datetime(dataframe.index).values.astype(np.int64) // precision_factor).astype(np.int64).astype(str)) # If tag columns exist, make an array of formatted tag keys and values diff --git a/influxdb/client.py b/influxdb/client.py index 51a64ac3..adab4edc 100644 --- a/influxdb/client.py +++ b/influxdb/client.py @@ -20,6 +20,7 @@ import msgpack import requests import requests.exceptions +from requests.adapters import HTTPAdapter from six.moves.urllib.parse import urlparse from influxdb.line_protocol import make_lines, quote_ident, quote_literal @@ -87,6 +88,11 @@ class InfluxDBClient(object): :param headers: headers to add to Requests, will add 'Content-Type' and 'Accept' unless these are already present, defaults to {} :type headers: dict + :param socket_options: use custom tcp socket options, + If not specified, then defaults are loaded from + ``HTTPConnection.default_socket_options`` + :type socket_options: list + :raises ValueError: if cert is provided but ssl is disabled (set to False) """ @@ -109,6 +115,7 @@ def __init__(self, gzip=False, session=None, headers=None, + socket_options=None, ): """Construct a new InfluxDBClient object.""" self.__host = host @@ -128,9 +135,10 @@ def __init__(self, session = requests.Session() self._session = session - adapter = requests.adapters.HTTPAdapter( + adapter = _SocketOptionsAdapter( pool_connections=int(pool_size), - pool_maxsize=int(pool_size) + pool_maxsize=int(pool_size), + socket_options=socket_options ) if use_udp: @@ -179,7 +187,7 @@ def __init__(self, def __enter__(self): """Enter function as used by context manager.""" - pass + return self def __exit__(self, _exc_type, _exc_value, _traceback): """Exit function as used by context manager.""" @@ -328,7 +336,10 @@ def request(self, url, method='GET', params=None, data=None, stream=False, _try = 0 while retry: try: - auth = (self._username, self._password) + if "Authorization" in headers: + auth = (None, None) + else: + auth = (self._username, self._password) response = self._session.request( method=method, url=url, @@ -626,7 +637,7 @@ def _batches(iterable, size): # http://code.activestate.com/recipes/303279-getting-items-in-batches/ iterator = iter(iterable) while True: - try: # Try get the first element in the iterator... + 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 @@ -1249,3 +1260,16 @@ def _msgpack_parse_hook(code, data): timestamp += datetime.timedelta(microseconds=(epoch_ns / 1000)) return timestamp.isoformat() + 'Z' return msgpack.ExtType(code, data) + + +class _SocketOptionsAdapter(HTTPAdapter): + """_SocketOptionsAdapter injects socket_options into HTTP Adapter.""" + + def __init__(self, *args, **kwargs): + self.socket_options = kwargs.pop("socket_options", None) + super(_SocketOptionsAdapter, self).__init__(*args, **kwargs) + + def init_poolmanager(self, *args, **kwargs): + if self.socket_options is not None: + kwargs["socket_options"] = self.socket_options + super(_SocketOptionsAdapter, self).init_poolmanager(*args, **kwargs) diff --git a/influxdb/tests/client_test.py b/influxdb/tests/client_test.py index e511ca9b..115fbc48 100644 --- a/influxdb/tests/client_test.py +++ b/influxdb/tests/client_test.py @@ -33,6 +33,7 @@ import requests_mock from nose.tools import raises +from urllib3.connection import HTTPConnection from influxdb import InfluxDBClient from influxdb.resultset import ResultSet @@ -1398,7 +1399,7 @@ def test_invalid_port_fails(self): InfluxDBClient('host', '80/redir', 'username', 'password') def test_chunked_response(self): - """Test chunked reponse for TestInfluxDBClient object.""" + """Test chunked response for TestInfluxDBClient object.""" example_response = \ u'{"results":[{"statement_id":0,"series":[{"columns":["key"],' \ '"values":[["cpu"],["memory"],["iops"],["network"]],"partial":' \ @@ -1498,6 +1499,40 @@ def test_auth_token(self): self.assertEqual(m.last_request.headers["Authorization"], "my-token") + def test_custom_socket_options(self): + """Test custom socket options.""" + test_socket_options = HTTPConnection.default_socket_options + \ + [(socket.SOL_SOCKET, socket.SO_KEEPALIVE, 1), + (socket.IPPROTO_TCP, socket.TCP_KEEPINTVL, 60), + (socket.IPPROTO_TCP, socket.TCP_KEEPCNT, 15)] + + cli = InfluxDBClient(username=None, password=None, + socket_options=test_socket_options) + + self.assertEquals(cli._session.adapters.get("http://").socket_options, + test_socket_options) + self.assertEquals(cli._session.adapters.get("http://").poolmanager. + connection_pool_kw.get("socket_options"), + test_socket_options) + + connection_pool = cli._session.adapters.get("http://").poolmanager \ + .connection_from_url( + url="http://localhost:8086") + new_connection = connection_pool._new_conn() + self.assertEquals(new_connection.socket_options, test_socket_options) + + def test_none_socket_options(self): + """Test default socket options.""" + cli = InfluxDBClient(username=None, password=None) + self.assertEquals(cli._session.adapters.get("http://").socket_options, + None) + connection_pool = cli._session.adapters.get("http://").poolmanager \ + .connection_from_url( + url="http://localhost:8086") + new_connection = connection_pool._new_conn() + self.assertEquals(new_connection.socket_options, + HTTPConnection.default_socket_options) + class FakeClient(InfluxDBClient): """Set up a fake client instance of InfluxDBClient.""" diff --git a/influxdb/tests/dataframe_client_test.py b/influxdb/tests/dataframe_client_test.py index f6db3c22..87b8e0d8 100644 --- a/influxdb/tests/dataframe_client_test.py +++ b/influxdb/tests/dataframe_client_test.py @@ -877,7 +877,7 @@ def test_query_into_dataframe(self): {"measurement": "network", "tags": {"direction": ""}, "columns": ["time", "value"], - "values":[["2009-11-10T23:00:00Z", 23422]] + "values": [["2009-11-10T23:00:00Z", 23422]] }, {"measurement": "network", "tags": {"direction": "in"}, @@ -1274,3 +1274,75 @@ def test_query_custom_index(self): self.assertListEqual(["time", "host"], list(_data_frame.index.names)) + + def test_dataframe_nanosecond_precision(self): + """Test nanosecond precision.""" + for_df_dict = { + "nanFloats": [1.1, float('nan'), 3.3, 4.4], + "onlyFloats": [1.1, 2.2, 3.3, 4.4], + "strings": ['one_one', 'two_two', 'three_three', 'four_four'] + } + df = pd.DataFrame.from_dict(for_df_dict) + df['time'] = ['2019-10-04 06:27:19.850557111+00:00', + '2019-10-04 06:27:19.850557184+00:00', + '2019-10-04 06:27:42.251396864+00:00', + '2019-10-04 06:27:42.251396974+00:00'] + df['time'] = pd.to_datetime(df['time'], unit='ns') + df = df.set_index('time') + + expected = ( + b'foo nanFloats=1.1,onlyFloats=1.1,strings="one_one" 1570170439850557111\n' # noqa E501 line too long + b'foo onlyFloats=2.2,strings="two_two" 1570170439850557184\n' # noqa E501 line too long + b'foo nanFloats=3.3,onlyFloats=3.3,strings="three_three" 1570170462251396864\n' # noqa E501 line too long + b'foo nanFloats=4.4,onlyFloats=4.4,strings="four_four" 1570170462251396974\n' # noqa E501 line too long + ) + + 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(df, 'foo', time_precision='n') + + self.assertEqual(m.last_request.body, expected) + + def test_dataframe_nanosecond_precision_one_microsecond(self): + """Test nanosecond precision within one microsecond.""" + # 1 microsecond = 1000 nanoseconds + start = np.datetime64('2019-10-04T06:27:19.850557000') + end = np.datetime64('2019-10-04T06:27:19.850558000') + + # generate timestamps with nanosecond precision + timestamps = np.arange( + start, + end + np.timedelta64(1, 'ns'), + np.timedelta64(1, 'ns') + ) + # generate values + values = np.arange(0.0, len(timestamps)) + + df = pd.DataFrame({'value': values}, index=timestamps) + 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(df, 'foo', time_precision='n') + + lines = m.last_request.body.decode('utf-8').split('\n') + self.assertEqual(len(lines), 1002) + + for index, line in enumerate(lines): + if index == 1001: + self.assertEqual(line, '') + continue + self.assertEqual( + line, + f"foo value={index}.0 157017043985055{7000 + index:04}" + )