diff --git a/CHANGELOG.md b/CHANGELOG.md index 8aa8de84..bfd27d38 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -4,14 +4,53 @@ 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.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) +- Add support for custom indexes for query in the DataFrameClient (#785) + +### Changed +- Amend retry to avoid sleep after last retry before raising exception (#790 thx @krzysbaranski) +- Remove msgpack pinning for requirements (#818 thx @prometheanfire) +- Update support for HTTP headers in the InfluxDBClient (#851 thx @bednar) + +### Removed + +## [v5.3.0] - 2020-04-10 ### Added - 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) +- 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) +- 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) - 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) +- Update repr of float values including properly handling of boolean (#488 thx @ghost) +- 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) +- 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 @@ -23,7 +62,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 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/README.rst b/README.rst index a40ed148..048db045 100644 --- a/README.rst +++ b/README.rst @@ -15,36 +15,45 @@ InfluxDB-Python :target: https://pypi.python.org/pypi/influxdb :alt: PyPI Status -InfluxDB-Python is a client for interacting with InfluxDB_. -Development of this library is maintained by: +.. important:: -+-----------+-------------------------------+ -| Github ID | URL | -+===========+===============================+ -| @aviau | (https://github.com/aviau) | -+-----------+-------------------------------+ -| @xginn8 | (https://github.com/xginn8) | -+-----------+-------------------------------+ -| @sebito91 | (https://github.com/sebito91) | -+-----------+-------------------------------+ + **This project is no longer in development** + + This v1 client library is for interacting with `InfluxDB 1.x `_ and 1.x-compatible endpoints in `InfluxDB 2.x `_. + Use it to: + + - Write data in line protocol. + - Query data with `InfluxQL `_. + + If you use `InfluxDB 2.x (TSM storage engine) `_ and `Flux `_, see the `v2 client library `_. + + If you use `InfluxDB 3.0 `_, see the `v3 client library `_. + + For new projects, consider using InfluxDB 3.0 and v3 client libraries. + +Description +=========== + +InfluxDB-python, the InfluxDB Python Client (1.x), is a client library for interacting with `InfluxDB 1.x `_ instances. .. _readme-about: -InfluxDB is an open-source distributed time series database, find more about InfluxDB_ at https://docs.influxdata.com/influxdb/latest +`InfluxDB`_ is the time series platform designed to handle high write and query loads. .. _installation: -InfluxDB pre v1.1.0 users -------------------------- -This module is tested with InfluxDB versions: v1.2.4, v1.3.9, v1.4.3, v1.5.4, v1.6.4, and 1.7.4. +For InfluxDB pre-v1.1.0 users +----------------------------- -Those users still on InfluxDB v0.8.x users may still use the legacy client by importing ``from influxdb.influxdb08 import InfluxDBClient``. +This module is tested with InfluxDB versions v1.2.4, v1.3.9, v1.4.3, v1.5.4, v1.6.4, and 1.7.4. -Installation ------------- +Users on InfluxDB v0.8.x may still use the legacy client by importing ``from influxdb.influxdb08 import InfluxDBClient``. + +For InfluxDB v1.1+ users +------------------------ Install, upgrade and uninstall influxdb-python with these commands:: @@ -152,21 +161,33 @@ We are also lurking on the following: Development ----------- +The v1 client libraries for InfluxDB 1.x were typically developed and maintained by InfluxDB community members. If you are an InfluxDB v1 user interested in maintaining this client library (at a minimum, keeping it updated with security patches) please contact the InfluxDB team at on the `Community Forums `_ or +`InfluxData Slack `_. + All development is done on Github_. Use Issues_ to report problems or submit contributions. .. _Github: https://github.com/influxdb/influxdb-python/ .. _Issues: https://github.com/influxdb/influxdb-python/issues -Please note that we WILL get to your questions/issues/concerns as quickly as possible. We maintain many -software repositories and sometimes things may get pushed to the backburner. Please don't take offense, -we will do our best to reply as soon as possible! +Please note that we will answer you question as quickly as possible. + +Maintainers: ++-----------+-------------------------------+ +| Github ID | URL | ++===========+===============================+ +| @aviau | (https://github.com/aviau) | ++-----------+-------------------------------+ +| @xginn8 | (https://github.com/xginn8) | ++-----------+-------------------------------+ +| @sebito91 | (https://github.com/sebito91) | ++-----------+-------------------------------+ Source code ----------- -The source code is currently available on Github: https://github.com/influxdata/influxdb-python +The source code for the InfluxDB Python Client (1.x) is currently available on Github: https://github.com/influxdata/influxdb-python TODO @@ -175,6 +196,6 @@ TODO The TODO/Roadmap can be found in Github bug tracker: https://github.com/influxdata/influxdb-python/issues -.. _InfluxDB: https://influxdata.com/time-series-platform/influxdb/ +.. _InfluxDB: https://influxdata.com/ .. _Sphinx: http://sphinx.pocoo.org/ .. _Tox: https://tox.readthedocs.org diff --git a/docs/source/conf.py b/docs/source/conf.py index 231c776c..efc22f88 100644 --- a/docs/source/conf.py +++ b/docs/source/conf.py @@ -117,7 +117,8 @@ # Add any paths that contain custom themes here, relative to this directory. #html_theme_path = [] -html_theme_path = [sphinx_rtd_theme.get_html_theme_path()] +# Calling get_html_theme_path is deprecated. +# html_theme_path = [sphinx_rtd_theme.get_html_theme_path()] # The name for this set of Sphinx documents. If None, it defaults to # " v documentation". diff --git a/docs/source/examples.rst b/docs/source/examples.rst index fdda62a9..841ad8b1 100644 --- a/docs/source/examples.rst +++ b/docs/source/examples.rst @@ -31,3 +31,9 @@ Tutorials - UDP .. literalinclude:: ../../examples/tutorial_udp.py :language: python + +Tutorials - Authorization by Token +================================== + +.. literalinclude:: ../../examples/tutorial_authorization.py + :language: python diff --git a/examples/tutorial_authorization.py b/examples/tutorial_authorization.py new file mode 100644 index 00000000..9d9a800f --- /dev/null +++ b/examples/tutorial_authorization.py @@ -0,0 +1,32 @@ +# -*- coding: utf-8 -*- +"""Tutorial how to authorize InfluxDB client by custom Authorization token.""" + +import argparse +from influxdb import InfluxDBClient + + +def main(token='my-token'): + """Instantiate a connection to the InfluxDB.""" + client = InfluxDBClient(username=None, password=None, + headers={"Authorization": token}) + + print("Use authorization token: " + token) + + version = client.ping() + print("Successfully connected to InfluxDB: " + version) + pass + + +def parse_args(): + """Parse the args from main.""" + parser = argparse.ArgumentParser( + description='example code to play with InfluxDB') + parser.add_argument('--token', type=str, required=False, + default='my-token', + help='Authorization token for the proxy that is ahead the InfluxDB.') + return parser.parse_args() + + +if __name__ == '__main__': + args = parse_args() + main(token=args.token) 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" }] } diff --git a/influxdb/__init__.py b/influxdb/__init__.py index b31170bb..e66f80ea 100644 --- a/influxdb/__init__.py +++ b/influxdb/__init__.py @@ -18,4 +18,4 @@ ] -__version__ = '5.2.3' +__version__ = '5.3.2' diff --git a/influxdb/_dataframe_client.py b/influxdb/_dataframe_client.py index d16e29ca..907db2cb 100644 --- a/influxdb/_dataframe_client.py +++ b/influxdb/_dataframe_client.py @@ -59,6 +59,8 @@ def write_points(self, :param dataframe: data points in a DataFrame :param measurement: name of measurement :param tags: dictionary of tags, with string key-values + :param tag_columns: [Optional, default None] List of data tag names + :param field_columns: [Options, default None] List of data field names :param time_precision: [Optional, default None] Either 's', 'ms', 'u' or 'n'. :param batch_size: [Optional] Value to write the points in batches @@ -150,7 +152,8 @@ def query(self, chunked=False, chunk_size=0, method="GET", - dropna=True): + dropna=True, + data_frame_index=None): """ Query data into a DataFrame. @@ -179,6 +182,8 @@ def query(self, containing all results within that chunk :param chunk_size: Size of each chunk to tell InfluxDB to use. :param dropna: drop columns where all values are missing + :param data_frame_index: the list of columns that + are used as DataFrame index :returns: the queried data :rtype: :class:`~.ResultSet` """ @@ -194,16 +199,18 @@ def query(self, results = super(DataFrameClient, self).query(query, **query_args) if query.strip().upper().startswith("SELECT"): if len(results) > 0: - return self._to_dataframe(results, dropna) + return self._to_dataframe(results, dropna, + data_frame_index=data_frame_index) else: return {} else: return results - def _to_dataframe(self, rs, dropna=True): + def _to_dataframe(self, rs, dropna=True, data_frame_index=None): 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 @@ -213,10 +220,15 @@ def _to_dataframe(self, rs, dropna=True): key = (name, tuple(sorted(tags.items()))) df = pd.DataFrame(data) df.time = pd.to_datetime(df.time) - df.set_index('time', inplace=True) - if df.index.tzinfo is None: - df.index = df.index.tz_localize('UTC') - df.index.name = None + + if data_frame_index: + df.set_index(data_frame_index, inplace=True) + else: + df.set_index('time', inplace=True) + if df.index.tzinfo is None: + df.index = df.index.tz_localize('UTC') + df.index.name = None + result[key].append(df) for key, data in result.items(): df = pd.concat(data).sort_index() @@ -251,7 +263,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') @@ -270,14 +283,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 @@ -342,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 @@ -371,7 +401,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) @@ -379,21 +410,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/client.py b/influxdb/client.py index ad4c6b66..c535a3f1 100644 --- a/influxdb/client.py +++ b/influxdb/client.py @@ -6,14 +6,21 @@ from __future__ import print_function from __future__ import unicode_literals -import time -import random - +import datetime +import gzip +import itertools +import io import json +import random import socket +import struct +import time +from itertools import chain, islice + +import msgpack import requests import requests.exceptions -from six.moves import xrange +from requests.adapters import HTTPAdapter from six.moves.urllib.parse import urlparse from influxdb.line_protocol import make_lines, quote_ident, quote_literal @@ -29,6 +36,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 @@ -50,8 +60,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 @@ -66,6 +80,18 @@ 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 + :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) """ @@ -86,6 +112,10 @@ def __init__(self, pool_size=10, path='', cert=None, + gzip=False, + session=None, + headers=None, + socket_options=None, ): """Construct a new InfluxDBClient object.""" self.__host = host @@ -99,11 +129,16 @@ def __init__(self, self._verify_ssl = verify_ssl self.__use_udp = use_udp - self.__udp_port = udp_port - self._session = requests.Session() - adapter = requests.adapters.HTTPAdapter( + self.__udp_port = int(udp_port) + + if not session: + session = requests.Session() + + self._session = session + adapter = _SocketOptionsAdapter( pool_connections=int(pool_size), - pool_maxsize=int(pool_size) + pool_maxsize=int(pool_size), + socket_options=socket_options ) if use_udp: @@ -142,10 +177,21 @@ def __init__(self, self._port, self._path) - self._headers = { - 'Content-Type': 'application/json', - 'Accept': 'text/plain' - } + if headers is None: + headers = {} + headers.setdefault('Content-Type', 'application/json') + headers.setdefault('Accept', 'application/x-msgpack') + self._headers = headers + + self._gzip = gzip + + def __enter__(self): + """Enter function as used by context manager.""" + return self + + def __exit__(self, _exc_type, _exc_value, _traceback): + """Exit function as used by context manager.""" + self.close() @property def _baseurl(self): @@ -231,7 +277,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. @@ -243,6 +289,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 @@ -266,17 +314,39 @@ 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 while retry: try: + if "Authorization" in headers: + auth = (None, None) + else: + auth = (self._username, self._password) response = self._session.request( method=method, url=url, - auth=(self._username, self._password), + auth=auth if None not in auth else None, params=params, data=data, + stream=stream, headers=headers, proxies=self._proxies, verify=self._verify_ssl, @@ -289,17 +359,34 @@ def request(self, url, method='GET', params=None, data=None, _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: + response._msgpack = msgpack.unpackb( + packed=response.content, + ext_hook=_msgpack_parse_hook, + raw=False) + else: + response._msgpack = None + + def reformat_error(response): + if response._msgpack: + return json.dumps(response._msgpack, separators=(',', ':')) + else: + return response.content + # if there's not an error, there must have been a successful response if 500 <= response.status_code < 600: - raise InfluxDBServerError(response.content) + raise InfluxDBServerError(reformat_error(response)) elif response.status_code == expected_response_code: return response else: - raise InfluxDBClientError(response.content, response.status_code) + err_msg = reformat_error(response) + raise InfluxDBClientError(err_msg, response.status_code) def write(self, data, params=None, expected_response_code=204, protocol='json'): @@ -308,7 +395,7 @@ def write(self, data, params=None, expected_response_code=204, :param data: the data to be written :type data: (if protocol is 'json') dict (if protocol is 'line') sequence of line protocol strings - or single string + or single string :param params: additional parameters for the request, defaults to None :type params: dict :param expected_response_code: the expected response code of the write @@ -319,7 +406,7 @@ def write(self, data, params=None, expected_response_code=204, :returns: True, if the write operation is successful :rtype: bool """ - headers = self._headers + headers = self._headers.copy() headers['Content-Type'] = 'application/octet-stream' if params: @@ -346,17 +433,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, @@ -447,13 +534,15 @@ def query(self, method=method, params=params, data=None, + stream=chunked, expected_response_code=expected_response_code ) - if chunked: - return self._read_chunked_response(response) - - data = response.json() + data = response._msgpack + if not data: + if chunked: + return self._read_chunked_response(response) + data = response.json() results = [ ResultSet(result, raise_errors=raise_errors) @@ -482,8 +571,9 @@ def write_points(self, :param points: the list of points to be written in the database :type points: list of dictionaries, each dictionary represents a point :type points: (if protocol is 'json') list of dicts, where each dict - represents a point. - (if protocol is 'line') sequence of line protocol strings. + represents a point. + (if protocol is 'line') sequence of line protocol strings. + :param time_precision: Either 's', 'm', 'ms' or 'u', defaults to None :type time_precision: str :param database: the database to write the points to. Defaults to @@ -544,8 +634,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, @@ -616,6 +715,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. @@ -740,7 +873,7 @@ def alter_retention_policy(self, name, database=None, query_string = ( "ALTER RETENTION POLICY {0} ON {1}" ).format(quote_ident(name), - quote_ident(database or self._database), shard_duration) + quote_ident(database or self._database)) if duration: query_string += " DURATION {0}".format(duration) if shard_duration: @@ -838,7 +971,7 @@ def drop_user(self, username): :param username: the username to drop :type username: str """ - text = "DROP USER {0}".format(quote_ident(username), method="POST") + text = "DROP USER {0}".format(quote_ident(username)) self.query(text, method="POST") def set_user_password(self, username, password): @@ -1119,3 +1252,25 @@ def _parse_netloc(netloc): 'password': info.password or None, 'host': info.hostname or 'localhost', 'port': info.port or 8086} + + +def _msgpack_parse_hook(code, data): + if code == 5: + (epoch_s, epoch_ns) = struct.unpack(">QI", data) + timestamp = datetime.datetime.utcfromtimestamp(epoch_s) + 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/helper.py b/influxdb/helper.py index e622526d..138cf6e8 100644 --- a/influxdb/helper.py +++ b/influxdb/helper.py @@ -41,6 +41,12 @@ 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 + # You should also make sure time is set in the given precision """ @@ -71,6 +77,15 @@ 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 {}, 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: @@ -116,11 +131,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 +158,12 @@ 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, + retention_policy=cls._retention_policy) + # will be None if not set and will default to ns cls._reset_() return rtn @@ -154,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/client.py b/influxdb/influxdb08/client.py index 965a91db..40c58145 100644 --- a/influxdb/influxdb08/client.py +++ b/influxdb/influxdb08/client.py @@ -292,10 +292,10 @@ def write_points(self, data, time_precision='s', *args, **kwargs): :type batch_size: int """ - def list_chunks(l, n): + def list_chunks(data_list, n): """Yield successive n-sized chunks from l.""" - for i in xrange(0, len(l), n): - yield l[i:i + n] + for i in xrange(0, len(data_list), n): + yield data_list[i:i + n] batch_size = kwargs.get('batch_size') if batch_size and batch_size > 0: 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/line_protocol.py b/influxdb/line_protocol.py index 249511d3..25dd2ad7 100644 --- a/influxdb/line_protocol.py +++ b/influxdb/line_protocol.py @@ -11,11 +11,19 @@ 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)) +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,19 +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 - elif precision == 'u': - return ns / 1e3 - elif precision == 'ms': - return ns / 1e6 - elif precision == 's': - return ns / 1e9 - elif precision == 'm': - return ns / 1e9 / 60 - elif precision == 'h': - return ns / 1e9 / 3600 + + if precision == 'u': + return ns / 10**3 + + if precision == 'ms': + return ns / 10**6 + + if precision == 's': + return ns / 10**9 + + if precision == 'm': + return ns / 10**9 / 60 + + if precision == 'h': + return ns / 10**9 / 3600 raise ValueError(timestamp) @@ -91,14 +104,21 @@ 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) - 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): - return repr(value) + + if isinstance(value, bool): + return str(value) + + if _is_float(value): + return repr(float(value)) return str(value) @@ -107,15 +127,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 +192,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/client_test.py b/influxdb/tests/client_test.py index 571b7ebc..115fbc48 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 @@ -31,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 @@ -214,6 +217,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: @@ -265,6 +333,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) @@ -473,6 +571,29 @@ def test_query(self): [{'value': 0.64, 'time': '2009-11-10T23:00:00Z'}] ) + def test_query_msgpack(self): + """Test query method with a messagepack response.""" + example_response = bytes(bytearray.fromhex( + "81a7726573756c74739182ac73746174656d656e745f696400a673657269" + "65739183a46e616d65a161a7636f6c756d6e7392a474696d65a176a67661" + "6c7565739192c70c05000000005d26178a019096c8cb3ff0000000000000" + )) + + with requests_mock.Mocker() as m: + m.register_uri( + requests_mock.GET, + "http://localhost:8086/query", + request_headers={"Accept": "application/x-msgpack"}, + headers={"Content-Type": "application/x-msgpack"}, + content=example_response + ) + rs = self.cli.query('select * from a') + + self.assertListEqual( + list(rs.get_points()), + [{'v': 1.0, 'time': '2019-07-10T16:51:22.026253Z'}] + ) + def test_select_into_post(self): """Test SELECT.*INTO is POSTed.""" example_response = ( @@ -666,6 +787,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":[{}]}' @@ -1078,7 +1259,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'], @@ -1218,18 +1399,13 @@ 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":' \ - '[{"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( @@ -1237,23 +1413,125 @@ 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__()) + + def test_auth_default(self): + """Test auth with default settings.""" + with requests_mock.Mocker() as m: + m.register_uri( + requests_mock.GET, + "http://localhost:8086/ping", + status_code=204, + headers={'X-Influxdb-Version': '1.2.3'} + ) + + cli = InfluxDBClient() + cli.ping() + + self.assertEqual(m.last_request.headers["Authorization"], + "Basic cm9vdDpyb290") + + def test_auth_username_password(self): + """Test auth with custom username and password.""" + with requests_mock.Mocker() as m: + m.register_uri( + requests_mock.GET, + "http://localhost:8086/ping", + status_code=204, + headers={'X-Influxdb-Version': '1.2.3'} + ) + + cli = InfluxDBClient(username='my-username', + password='my-password') + cli.ping() + + self.assertEqual(m.last_request.headers["Authorization"], + "Basic bXktdXNlcm5hbWU6bXktcGFzc3dvcmQ=") + + def test_auth_username_password_none(self): + """Test auth with not defined username or password.""" + with requests_mock.Mocker() as m: + m.register_uri( + requests_mock.GET, + "http://localhost:8086/ping", + status_code=204, + headers={'X-Influxdb-Version': '1.2.3'} + ) + + cli = InfluxDBClient(username=None, password=None) + cli.ping() + self.assertFalse('Authorization' in m.last_request.headers) + + cli = InfluxDBClient(username=None) + cli.ping() + self.assertFalse('Authorization' in m.last_request.headers) + + cli = InfluxDBClient(password=None) + cli.ping() + self.assertFalse('Authorization' in m.last_request.headers) + + def test_auth_token(self): + """Test auth with custom authorization header.""" + with requests_mock.Mocker() as m: + m.register_uri( + requests_mock.GET, + "http://localhost:8086/ping", + status_code=204, + headers={'X-Influxdb-Version': '1.2.3'} + ) + + cli = InfluxDBClient(username=None, password=None, + headers={"Authorization": "my-token"}) + cli.ping() + 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): diff --git a/influxdb/tests/dataframe_client_test.py b/influxdb/tests/dataframe_client_test.py index 90312ed8..87b8e0d8 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' @@ -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"}, @@ -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, np.NaN], + ["2015-01-29T21:55:43.702900257Z", + 23422, 122878, np.NaN], + ["2015-06-11T20:46:02Z", + 0.64, 0.5434, np.NaN] + ] + } + ] + }, { + "series": [ + { + "name": "cpu_load_short", + "columns": ["time", "count"], + "values": [ + ["1970-01-01T00:00:00Z", 3] + ] + } + ] + } + ] + } + + pd1 = pd.DataFrame( + [[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", + "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') @@ -1032,3 +1124,225 @@ 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) + + def test_query_custom_index(self): + """Test query with custom indexes.""" + data = { + "results": [ + { + "series": [ + { + "name": "cpu_load_short", + "columns": ["time", "value", "host"], + "values": [ + [1, 0.55, "local"], + [2, 23422, "local"], + [3, 0.64, "local"] + ] + } + ] + } + ] + } + + 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'} + with _mocked_session(cli, 'GET', 200, data): + result = cli.query(iql, bind_params=bind_params, + data_frame_index=["time", "host"]) + + _data_frame = result['cpu_load_short'] + print(_data_frame) + + 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}" + ) diff --git a/influxdb/tests/helper_test.py b/influxdb/tests/helper_test.py index 6f24e85d..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() @@ -310,8 +318,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'}) @@ -365,3 +384,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) 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 fda3f720..a0263243 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) @@ -450,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)) @@ -817,6 +846,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): @@ -855,3 +942,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 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 diff --git a/influxdb/tests/test_line_protocol.py b/influxdb/tests/test_line_protocol.py index bccd7727..5b344990 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 @@ -42,7 +44,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' ) @@ -115,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 = { @@ -166,3 +186,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' + ) diff --git a/pyproject.toml b/pyproject.toml new file mode 100644 index 00000000..1b68d94e --- /dev/null +++ b/pyproject.toml @@ -0,0 +1,3 @@ +[build-system] +requires = ["setuptools>=42", "wheel"] +build-backend = "setuptools.build_meta" \ No newline at end of file diff --git a/requirements.txt b/requirements.txt index db5f6f85..a3df3154 100644 --- a/requirements.txt +++ b/requirements.txt @@ -1,4 +1,5 @@ python-dateutil>=2.6.0 -pytz +pytz>=2016.10 requests>=2.17.0 six>=1.10.0 +msgpack>=0.5.0 diff --git a/setup.py b/setup.py index d44875f6..8ac7d1a7 100755 --- a/setup.py +++ b/setup.py @@ -23,6 +23,11 @@ with open('requirements.txt', 'r') as f: requires = [x.strip() for x in f if x.strip()] +# Debugging: Print the requires values +print("install_requires values:") +for req in requires: + print(f"- {req}") + with open('test-requirements.txt', 'r') as f: test_requires = [x.strip() for x in f if x.strip()] diff --git a/tox.ini b/tox.ini index ff30ebac..a1005abb 100644 --- a/tox.ini +++ b/tox.ini @@ -12,8 +12,8 @@ deps = -r{toxinidir}/requirements.txt py35: numpy==1.14.6 py36: pandas==0.23.4 py36: numpy==1.15.4 - py37: pandas==0.24.2 - py37: numpy==1.16.2 + py37: pandas>=0.24.2 + py37: numpy>=1.16.2 # Only install pandas with non-pypy interpreters # Testing all combinations would be too expensive commands = nosetests -v --with-doctest {posargs} @@ -31,16 +31,16 @@ commands = pydocstyle --count -ve examples influxdb [testenv:coverage] deps = -r{toxinidir}/requirements.txt -r{toxinidir}/test-requirements.txt - pandas + pandas==0.24.2 coverage numpy commands = nosetests -v --with-coverage --cover-html --cover-package=influxdb [testenv:docs] deps = -r{toxinidir}/requirements.txt - pandas==0.24.2 - numpy==1.16.2 - Sphinx==1.8.5 + pandas>=0.24.2 + numpy>=1.16.2 + Sphinx>=1.8.5 sphinx_rtd_theme commands = sphinx-build -b html docs/source docs/build