PK 0:
return self._to_dataframe(results)
else:
return {}
else:
return results
def get_list_series(self, database=None):
"""
Get the list of series, in DataFrame
"""
results = super(DataFrameClient, self)\
.query("SHOW SERIES", database=database)
if len(results):
return dict(
(key[0], pd.DataFrame(data)) for key, data in results.items()
)
else:
return {}
def _to_dataframe(self, rs):
result = {}
for key, data in rs.items():
name, tags = key
if tags is None:
key = name
else:
key = (name, tuple(sorted(tags.items())))
df = pd.DataFrame(data)
df.time = pd.to_datetime(df.time)
df.set_index('time', inplace=True)
df.index = df.index.tz_localize('UTC')
df.index.name = None
result[key] = df
return result
def _convert_dataframe_to_json(self, dataframe, measurement, tags=None,
time_precision=None):
if not isinstance(dataframe, pd.DataFrame):
raise TypeError('Must be DataFrame, but type was: {}.'
.format(type(dataframe)))
if not (isinstance(dataframe.index, pd.tseries.period.PeriodIndex) or
isinstance(dataframe.index, pd.tseries.index.DatetimeIndex)):
raise TypeError('Must be DataFrame with DatetimeIndex or \
PeriodIndex.')
dataframe.index = dataframe.index.to_datetime()
if dataframe.index.tzinfo is None:
dataframe.index = dataframe.index.tz_localize('UTC')
# Convert column to strings
dataframe.columns = dataframe.columns.astype('str')
# Convert dtype for json serialization
dataframe = dataframe.astype('object')
precision_factor = {
"n": 1,
"u": 1e3,
"ms": 1e6,
"s": 1e9,
"m": 1e9 * 60,
"h": 1e9 * 3600,
}.get(time_precision, 1)
points = [
{'measurement': measurement,
'tags': tags if tags else {},
'fields': rec,
'time': int(ts.value / precision_factor)
}
for ts, rec in zip(dataframe.index, dataframe.to_dict('record'))]
return points
def _datetime_to_epoch(self, datetime, time_precision='s'):
seconds = (datetime - self.EPOCH).total_seconds()
if time_precision == 'h':
return seconds / 3600
elif time_precision == 'm':
return seconds / 60
elif time_precision == 's':
return seconds
elif time_precision == 'ms':
return seconds * 1e3
elif time_precision == 'u':
return seconds * 1e6
elif time_precision == 'n':
return seconds * 1e9
PK
= cls._bulk_size:
cls.commit()
@classmethod
def commit(cls, client=None):
"""
Commit everything from datapoints via the client.
:param client: InfluxDBClient instance for writing points to InfluxDB.
:attention: any provided client will supersede the class client.
:return: result of client.write_points.
"""
if not client:
client = cls._client
rtn = client.write_points(cls._json_body_())
cls._reset_()
return rtn
@classmethod
def _json_body_(cls):
"""
:return: JSON body of these datapoints.
"""
json = []
for series_name, data in six.iteritems(cls._datapoints):
for point in data:
json_point = {
"measurement": series_name,
"fields": {},
"tags": {},
}
for field in cls._fields:
json_point['fields'][field] = point.__dict__[field]
for tag in cls._tags:
json_point['tags'][tag] = point.__dict__[tag]
json.append(json_point)
return json
@classmethod
def _reset_(cls):
"""
Reset data storage.
"""
cls._datapoints = defaultdict(list)
PK
# Source: https://gist.github.com/sampsyo/920215
#
import json
_decoder = json.JSONDecoder()
def loads(s):
"""A generator reading a sequence of JSON values from a string."""
while s:
s = s.strip()
obj, pos = _decoder.raw_decode(s)
if not pos:
raise ValueError('no JSON object found at %i' % pos)
yield obj
s = s[pos:]
PK
> cli = InfluxDBClient.from_DSN('influxdb://username:password@\
localhost:8086/databasename', timeout=5)
>> type(cli)
>> cli = InfluxDBClient.from_DSN('udp+influxdb://username:pass@\
localhost:8086/databasename', timeout=5, udp_port=159)
>> print('{0._baseurl} - {0.use_udp} {0.udp_port}'.format(cli))
http://localhost:8086 - True 159
.. note:: parameters provided in `**kwargs` may override dsn parameters
.. note:: when using "udp+influxdb" the specified port (if any) will
be used for the TCP connection; specify the UDP port with the
additional `udp_port` parameter (cf. examples).
"""
init_args = {}
conn_params = urlparse(dsn)
scheme_info = conn_params.scheme.split('+')
if len(scheme_info) == 1:
scheme = scheme_info[0]
modifier = None
else:
modifier, scheme = scheme_info
if scheme != 'influxdb':
raise ValueError('Unknown scheme "{}".'.format(scheme))
if modifier:
if modifier == 'udp':
init_args['use_udp'] = True
elif modifier == 'https':
init_args['ssl'] = True
else:
raise ValueError('Unknown modifier "{}".'.format(modifier))
if conn_params.hostname:
init_args['host'] = conn_params.hostname
if conn_params.port:
init_args['port'] = conn_params.port
if conn_params.username:
init_args['username'] = conn_params.username
if conn_params.password:
init_args['password'] = conn_params.password
if conn_params.path and len(conn_params.path) > 1:
init_args['database'] = conn_params.path[1:]
init_args.update(kwargs)
return InfluxDBClient(**init_args)
def switch_database(self, database):
"""Change the client's database.
:param database: the name of the database to switch to
:type database: str
"""
self._database = database
def switch_user(self, username, password):
"""Change the client's username.
:param username: the username to switch to
:type username: str
:param password: the password for the username
:type password: str
"""
self._username = username
self._password = password
def request(self, url, method='GET', params=None, data=None,
expected_response_code=200, headers=None):
"""Make a HTTP request to the InfluxDB API.
:param url: the path of the HTTP request, e.g. write, query, etc.
:type url: str
:param method: the HTTP method for the request, defaults to GET
:type method: str
:param params: additional parameters for the request, defaults to None
:type params: dict
:param data: the data of the request, defaults to None
:type data: str
:param expected_response_code: the expected response code of
the request, defaults to 200
:type expected_response_code: int
:returns: the response from the request
:rtype: :class:`requests.Response`
:raises InfluxDBClientError: if the response code is not the
same as `expected_response_code`
"""
url = "{0}/{1}".format(self._baseurl, url)
if headers is None:
headers = self._headers
if params is None:
params = {}
if isinstance(data, (dict, list)):
data = json.dumps(data)
# Try to send the request a maximum of three times. (see #103)
# TODO (aviau): Make this configurable.
for i in range(0, 3):
try:
response = self._session.request(
method=method,
url=url,
auth=(self._username, self._password),
params=params,
data=data,
headers=headers,
verify=self._verify_ssl,
timeout=self._timeout
)
break
except requests.exceptions.ConnectionError as e:
if i < 2:
continue
else:
raise e
if response.status_code == expected_response_code:
return response
else:
raise InfluxDBClientError(response.content, response.status_code)
def write(self, data, params=None, expected_response_code=204):
"""Write data to InfluxDB.
:param data: the data to be written
:type data: dict
:param params: additional parameters for the request, defaults to None
:type params: dict
:param expected_response_code: the expected response code of the write
operation, defaults to 204
:type expected_response_code: int
:returns: True, if the write operation is successful
:rtype: bool
"""
headers = self._headers
headers['Content-type'] = 'application/octet-stream'
if params:
precision = params.get('precision')
else:
precision = None
self.request(
url="write",
method='POST',
params=params,
data=make_lines(data, precision).encode('utf-8'),
expected_response_code=expected_response_code,
headers=headers
)
return True
def query(self,
query,
params={},
epoch=None,
expected_response_code=200,
database=None,
raise_errors=True):
"""Send a query to InfluxDB.
:param query: the actual query string
:type query: str
:param params: additional parameters for the request, defaults to {}
:type params: dict
:param expected_response_code: the expected status code of response,
defaults to 200
:type expected_response_code: int
:param database: database to query, defaults to None
:type database: str
:param raise_errors: Whether or not to raise exceptions when InfluxDB
returns errors, defaults to True
:type raise_errors: bool
:returns: the queried data
:rtype: :class:`~.ResultSet`
"""
params['q'] = query
params['db'] = database or self._database
if epoch is not None:
params['epoch'] = epoch
response = self.request(
url="query",
method='GET',
params=params,
data=None,
expected_response_code=expected_response_code
)
data = response.json()
results = [
ResultSet(result, raise_errors=raise_errors)
for result
in data.get('results', [])
]
# TODO(aviau): Always return a list. (This would be a breaking change)
if len(results) == 1:
return results[0]
else:
return results
def write_points(self,
points,
time_precision=None,
database=None,
retention_policy=None,
tags=None,
batch_size=None,
):
"""Write to multiple time series names.
:param points: the list of points to be written in the database
:type points: list of dictionaries, each dictionary represents a point
: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
the client's current database
:type database: str
:param tags: a set of key-value pairs associated with each point. Both
keys and values must be strings. These are shared tags and will be
merged with point-specific tags, defaults to None
:type tags: dict
:param retention_policy: the retention policy for the points. Defaults
to None
:type retention_policy: str
:param batch_size: value to write the points in batches
instead of all at one time. Useful for when doing data dumps from
one database to another or when doing a massive write operation,
defaults to None
:type batch_size: int
:returns: True, if the operation is successful
:rtype: bool
.. note:: if no retention policy is specified, the default retention
policy for the database is used
"""
if batch_size and batch_size > 0:
for batch in self._batches(points, batch_size):
self._write_points(points=batch,
time_precision=time_precision,
database=database,
retention_policy=retention_policy,
tags=tags)
return True
else:
return self._write_points(points=points,
time_precision=time_precision,
database=database,
retention_policy=retention_policy,
tags=tags)
def _batches(self, iterable, size):
for i in xrange(0, len(iterable), size):
yield iterable[i:i + size]
def _write_points(self,
points,
time_precision,
database,
retention_policy,
tags):
if time_precision not in ['n', 'u', 'ms', 's', 'm', 'h', None]:
raise ValueError(
"Invalid time precision is given. "
"(use 'n', 'u', 'ms', 's', 'm' or 'h')")
if self.use_udp and time_precision and time_precision != 's':
raise ValueError(
"InfluxDB only supports seconds precision for udp writes"
)
data = {
'points': points
}
if tags is not None:
data['tags'] = tags
params = {
'db': database or self._database
}
if time_precision is not None:
params['precision'] = time_precision
if retention_policy is not None:
params['rp'] = retention_policy
if self.use_udp:
self.send_packet(data)
else:
self.write(
data=data,
params=params,
expected_response_code=204
)
return True
def get_list_database(self):
"""Get the list of databases in InfluxDB.
:returns: all databases in InfluxDB
:rtype: list of dictionaries
:Example:
::
>> dbs = client.get_list_database()
>> dbs
[{u'name': u'db1'}, {u'name': u'db2'}, {u'name': u'db3'}]
"""
return list(self.query("SHOW DATABASES").get_points())
def create_database(self, dbname):
"""Create a new database in InfluxDB.
:param dbname: the name of the database to create
:type dbname: str
"""
self.query("CREATE DATABASE %s" % dbname)
def drop_database(self, dbname):
"""Drop a database from InfluxDB.
:param dbname: the name of the database to drop
:type dbname: str
"""
self.query("DROP DATABASE %s" % dbname)
def create_retention_policy(self, name, duration, replication,
database=None, default=False):
"""Create a retention policy for a database.
:param name: the name of the new retention policy
:type name: str
:param duration: the duration of the new retention policy.
Durations such as 1h, 90m, 12h, 7d, and 4w, are all supported
and mean 1 hour, 90 minutes, 12 hours, 7 day, and 4 weeks,
respectively. For infinite retention – meaning the data will
never be deleted – use 'INF' for duration.
The minimum retention period is 1 hour.
:type duration: str
:param replication: the replication of the retention policy
:type replication: str
:param database: the database for which the retention policy is
created. Defaults to current client's database
:type database: str
:param default: whether or not to set the policy as default
:type default: bool
"""
query_string = \
"CREATE RETENTION POLICY %s ON %s " \
"DURATION %s REPLICATION %s" % \
(name, database or self._database, duration, replication)
if default is True:
query_string += " DEFAULT"
self.query(query_string)
def alter_retention_policy(self, name, database=None,
duration=None, replication=None, default=None):
"""Mofidy an existing retention policy for a database.
:param name: the name of the retention policy to modify
:type name: str
:param database: the database for which the retention policy is
modified. Defaults to current client's database
:type database: str
:param duration: the new duration of the existing retention policy.
Durations such as 1h, 90m, 12h, 7d, and 4w, are all supported
and mean 1 hour, 90 minutes, 12 hours, 7 day, and 4 weeks,
respectively. For infinite retention – meaning the data will
never be deleted – use 'INF' for duration.
The minimum retention period is 1 hour.
:type duration: str
:param replication: the new replication of the existing
retention policy
:type replication: str
:param default: whether or not to set the modified policy as default
:type default: bool
.. note:: at least one of duration, replication, or default flag
should be set. Otherwise the operation will fail.
"""
query_string = (
"ALTER RETENTION POLICY {} ON {}"
).format(name, database or self._database)
if duration:
query_string += " DURATION {}".format(duration)
if replication:
query_string += " REPLICATION {}".format(replication)
if default is True:
query_string += " DEFAULT"
self.query(query_string)
def get_list_retention_policies(self, database=None):
"""Get the list of retention policies for a database.
:param database: the name of the database, defaults to the client's
current database
:type database: str
:returns: all retention policies for the database
:rtype: list of dictionaries
:Example:
::
>> ret_policies = client.get_list_retention_policies('my_db')
>> ret_policies
[{u'default': True,
u'duration': u'0',
u'name': u'default',
u'replicaN': 1}]
"""
rsp = self.query(
"SHOW RETENTION POLICIES ON %s" % (database or self._database)
)
return list(rsp.get_points())
def get_list_series(self, database=None):
"""Get the list of series for a database.
:param database: the name of the database, defaults to the client's
current database
:type database: str
:returns: all series in the specified database
:rtype: list of dictionaries
:Example:
>> series = client.get_list_series('my_database')
>> series
[{'name': u'cpu_usage',
'tags': [{u'_id': 1,
u'host': u'server01',
u'region': u'us-west'}]}]
"""
rsp = self.query("SHOW SERIES", database=database)
series = []
for serie in rsp.items():
series.append(
{
"name": serie[0][0],
"tags": list(serie[1])
}
)
return series
def get_list_users(self):
"""Get the list of all users in InfluxDB.
:returns: all users in InfluxDB
:rtype: list of dictionaries
:Example:
::
>> users = client.get_list_users()
>> users
[{u'admin': True, u'user': u'user1'},
{u'admin': False, u'user': u'user2'},
{u'admin': False, u'user': u'user3'}]
"""
return list(self.query("SHOW USERS").get_points())
def create_user(self, username, password, admin=False):
"""Create a new user in InfluxDB
:param username: the new username to create
:type username: str
:param password: the password for the new user
:type password: str
:param admin: whether the user should have cluster administration
privileges or not
:type admin: boolean
"""
text = "CREATE USER {} WITH PASSWORD '{}'".format(username, password)
if admin:
text += ' WITH ALL PRIVILEGES'
self.query(text)
def drop_user(self, username):
"""Drop an user from InfluxDB.
:param username: the username to drop
:type username: str
"""
text = "DROP USER {}".format(username)
self.query(text)
def set_user_password(self, username, password):
"""Change the password of an existing user.
:param username: the username who's password is being changed
:type username: str
:param password: the new password for the user
:type password: str
"""
text = "SET PASSWORD FOR {} = '{}'".format(username, password)
self.query(text)
def delete_series(self, database=None, measurement=None, tags=None):
"""Delete series from a database. Series can be filtered by
measurement and tags.
:param measurement: Delete all series from a measurement
:type id: string
:param tags: Delete all series that match given tags
:type id: dict
:param database: the database from which the series should be
deleted, defaults to client's current database
:type database: str
"""
database = database or self._database
query_str = 'DROP SERIES'
if measurement:
query_str += ' FROM "{}"'.format(measurement)
if tags:
query_str += ' WHERE ' + ' and '.join(["{}='{}'".format(k, v)
for k, v in tags.items()])
self.query(query_str, database=database)
def revoke_admin_privileges(self, username):
"""Revoke cluster administration privileges from an user.
:param username: the username to revoke privileges from
:type username: str
.. note:: Only a cluster administrator can create/ drop databases
and manage users.
"""
text = "REVOKE ALL PRIVILEGES FROM {}".format(username)
self.query(text)
def grant_privilege(self, privilege, database, username):
"""Grant a privilege on a database to an user.
:param privilege: the privilege to grant, one of 'read', 'write'
or 'all'. The string is case-insensitive
:type privilege: str
:param database: the database to grant the privilege on
:type database: str
:param username: the username to grant the privilege to
:type username: str
"""
text = "GRANT {} ON {} TO {}".format(privilege,
database,
username)
self.query(text)
def revoke_privilege(self, privilege, database, username):
"""Revoke a privilege on a database from an user.
:param privilege: the privilege to revoke, one of 'read', 'write'
or 'all'. The string is case-insensitive
:type privilege: str
:param database: the database to revoke the privilege on
:type database: str
:param username: the username to revoke the privilege from
:type username: str
"""
text = "REVOKE {} ON {} FROM {}".format(privilege,
database,
username)
self.query(text)
def send_packet(self, packet):
"""Send an UDP packet.
:param packet: the packet to be sent
:type packet: dict
"""
data = make_lines(packet).encode('utf-8')
self.udp_socket.sendto(data, (self._host, self.udp_port))
class InfluxDBClusterClient(object):
"""The :class:`~.InfluxDBClusterClient` is the client for connecting
to a cluster of InfluxDB servers. It's basically a proxy to multiple
InfluxDBClients.
:param hosts: all hosts to be included in the cluster, each of which
should be in the format (address, port),
e.g. [('127.0.0.1', 8086), ('127.0.0.1', 9096)]. Defaults to
[('localhost', 8086)]
:type hosts: list of tuples
:param shuffle: whether the queries should hit servers evenly(randomly),
defaults to True
:type shuffle: bool
:param client_base_class: the base class for all clients in the cluster.
This parameter is used to enable the support of different client
types. Defaults to :class:`~.InfluxDBClient`
"""
def __init__(self,
hosts=[('localhost', 8086)],
username='root',
password='root',
database=None,
ssl=False,
verify_ssl=False,
timeout=None,
use_udp=False,
udp_port=4444,
shuffle=True,
client_base_class=InfluxDBClient,
):
self.clients = []
self.bad_clients = [] # Corresponding server has failures in history
self.shuffle = shuffle
for h in hosts:
self.clients.append(client_base_class(host=h[0], port=h[1],
username=username,
password=password,
database=database,
ssl=ssl,
verify_ssl=verify_ssl,
timeout=timeout,
use_udp=use_udp,
udp_port=udp_port))
for method in dir(client_base_class):
if method.startswith('_'):
continue
orig_func = getattr(client_base_class, method)
if not callable(orig_func):
continue
setattr(self, method, self._make_func(orig_func))
@staticmethod
def from_DSN(dsn, client_base_class=InfluxDBClient,
shuffle=True, **kwargs):
"""Same as :meth:`~.InfluxDBClient.from_DSN`, but supports
multiple servers.
:param shuffle: whether the queries should hit servers
evenly(randomly), defaults to True
:type shuffle: bool
:param client_base_class: the base class for all clients in the
cluster. This parameter is used to enable the support of
different client types. Defaults to :class:`~.InfluxDBClient`
:Example:
::
>> cluster = InfluxDBClusterClient.from_DSN('influxdb://usr:pwd\
@host1:8086,usr:pwd@host2:8086/db_name', timeout=5)
>> type(cluster)
>> cluster.clients
[,
]
"""
conn_params = urlparse(dsn)
netlocs = conn_params.netloc.split(',')
cluster_client = InfluxDBClusterClient(
hosts=[],
client_base_class=client_base_class,
shuffle=shuffle,
**kwargs)
for netloc in netlocs:
single_dsn = '%(scheme)s://%(netloc)s%(path)s' % (
{'scheme': conn_params.scheme,
'netloc': netloc,
'path': conn_params.path}
)
cluster_client.clients.append(client_base_class.from_DSN(
single_dsn,
**kwargs))
return cluster_client
def _make_func(self, orig_func):
@wraps(orig_func)
def func(*args, **kwargs):
if self.shuffle:
random.shuffle(self.clients)
clients = self.clients + self.bad_clients
for c in clients:
bad_client = False
try:
return orig_func(c, *args, **kwargs)
except InfluxDBClientError as e:
# Errors caused by user's requests, re-raise
raise e
except Exception as e:
# Errors that might caused by server failure, try another
bad_client = True
if c in self.clients:
self.clients.remove(c)
self.bad_clients.append(c)
finally:
if not bad_client and c in self.bad_clients:
self.bad_clients.remove(c)
self.clients.append(c)
raise InfluxDBServerError("InfluxDB: no viable server!")
return func
PK Started influxdb bin in %r with ports %s and %s.." % (
datetime.datetime.now(),
self.temp_dir_base,
self.admin_port,
self.http_port
)
)
# wait for it to listen on the broker and admin ports:
# usually a fresh instance is ready in less than 1 sec ..
timeout = time.time() + 10 # so 10 secs should be enough,
# otherwise either your system load is high,
# or you run a 286 @ 1Mhz ?
try:
while time.time() < timeout:
if (is_port_open(self.http_port)
and is_port_open(self.admin_port)):
# it's hard to check if a UDP port is open..
if udp_enabled:
# so let's just sleep 0.5 sec in this case
# to be sure that the server has open the port
time.sleep(0.5)
break
time.sleep(0.5)
if self.proc.poll() is not None:
raise RuntimeError('influxdb prematurely exited')
else:
self.proc.terminate()
self.proc.wait()
raise RuntimeError('Timeout waiting for influxdb to listen'
' on its ports (%s)' % ports)
except RuntimeError as err:
data = self.get_logs_and_output()
data['reason'] = str(err)
data['now'] = datetime.datetime.now()
raise RuntimeError("%(now)s > %(reason)s. RC=%(rc)s\n"
"stdout=%(out)s\nstderr=%(err)s\nlogs=%(logs)r"
% data)
def find_influxd_path(self):
influxdb_bin_path = os.environ.get(
'INFLUXDB_PYTHON_INFLUXD_PATH',
None
)
if influxdb_bin_path is None:
influxdb_bin_path = distutils.spawn.find_executable('influxd')
if not influxdb_bin_path:
try:
influxdb_bin_path = subprocess.check_output(
['which', 'influxdb']
).strip()
except subprocess.CalledProcessError:
# fallback on :
influxdb_bin_path = '/opt/influxdb/influxd'
if not os.path.isfile(influxdb_bin_path):
raise unittest.SkipTest("Could not find influxd binary")
version = subprocess.check_output([influxdb_bin_path, 'version'])
print("InfluxDB version: %s" % version, file=sys.stderr)
return influxdb_bin_path
def get_logs_and_output(self):
proc = self.proc
try:
with open(self.logs_file) as fh:
logs = fh.read()
except IOError as err:
logs = "Couldn't read logs: %s" % err
return {
'rc': proc.returncode,
'out': proc.stdout.read(),
'err': proc.stderr.read(),
'logs': logs
}
def close(self, remove_tree=True):
self.proc.terminate()
self.proc.wait()
if remove_tree:
shutil.rmtree(self.temp_dir_base)
PK
= cls._bulk_size:
cls.commit()
@classmethod
def commit(cls, client=None):
"""
Commit everything from datapoints via the client.
:param client: InfluxDBClient instance for writing points to InfluxDB.
:attention: any provided client will supersede the class client.
:return: result of client.write_points.
"""
if not client:
client = cls._client
rtn = client.write_points(cls._json_body_())
cls._reset_()
return rtn
@classmethod
def _json_body_(cls):
"""
:return: JSON body of these datapoints.
"""
json = []
for series_name, data in six.iteritems(cls._datapoints):
json.append({'name': series_name,
'columns': cls._fields,
'points': [[point.__dict__[k] for k in cls._fields]
for point in data]
})
return json
@classmethod
def _reset_(cls):
"""
Reset data storage.
"""
cls._datapoints = defaultdict(list)
PK
# Source: https://gist.github.com/sampsyo/920215
#
import json
_decoder = json.JSONDecoder()
def loads(s):
"""A generator reading a sequence of JSON values from a string."""
while s:
s = s.strip()
obj, pos = _decoder.raw_decode(s)
if not pos:
raise ValueError('no JSON object found at %i' % pos)
yield obj
s = s[pos:]
PK
> cli = InfluxDBClient.from_DSN('influxdb://username:password@\
... localhost:8086/databasename', timeout=5)
>> type(cli)
>> cli = InfluxDBClient.from_DSN('udp+influxdb://username:pass@\
... localhost:8086/databasename', timeout=5, udp_port=159)
>> print('{0._baseurl} - {0.use_udp} {0.udp_port}'.format(cli))
http://localhost:8086 - True 159
:param dsn: data source name
:type dsn: string
:param **kwargs: additional parameters for InfluxDBClient.
:type **kwargs: dict
:note: parameters provided in **kwargs may override dsn parameters.
:note: when using "udp+influxdb" the specified port (if any) will be
used for the TCP connection; specify the udp port with the additional
udp_port parameter (cf. examples).
:raise ValueError: if the provided DSN has any unexpected value.
"""
init_args = {}
conn_params = urlparse(dsn)
scheme_info = conn_params.scheme.split('+')
if len(scheme_info) == 1:
scheme = scheme_info[0]
modifier = None
else:
modifier, scheme = scheme_info
if scheme != 'influxdb':
raise ValueError('Unknown scheme "{}".'.format(scheme))
if modifier:
if modifier == 'udp':
init_args['use_udp'] = True
elif modifier == 'https':
init_args['ssl'] = True
else:
raise ValueError('Unknown modifier "{}".'.format(modifier))
if conn_params.hostname:
init_args['host'] = conn_params.hostname
if conn_params.port:
init_args['port'] = conn_params.port
if conn_params.username:
init_args['username'] = conn_params.username
if conn_params.password:
init_args['password'] = conn_params.password
if conn_params.path and len(conn_params.path) > 1:
init_args['database'] = conn_params.path[1:]
init_args.update(kwargs)
return InfluxDBClient(**init_args)
# Change member variables
def switch_database(self, database):
"""
switch_database()
Change client database.
:param database: the new database name to switch to
:type database: string
"""
self._database = database
def switch_db(self, database):
"""
DEPRECATED. Change client database.
"""
warnings.warn(
"switch_db is deprecated, and will be removed "
"in future versions. Please use "
"``InfluxDBClient.switch_database(database)`` instead.",
FutureWarning)
return self.switch_database(database)
def switch_user(self, username, password):
"""
switch_user()
Change client username.
:param username: the new username to switch to
:type username: string
:param password: the new password to switch to
:type password: string
"""
self._username = username
self._password = password
def request(self, url, method='GET', params=None, data=None,
expected_response_code=200):
"""
Make a http request to API
"""
url = "{0}/{1}".format(self._baseurl, url)
if params is None:
params = {}
auth = {
'u': self._username,
'p': self._password
}
params.update(auth)
if data is not None and not isinstance(data, str):
data = json.dumps(data)
# Try to send the request a maximum of three times. (see #103)
# TODO (aviau): Make this configurable.
for i in range(0, 3):
try:
response = session.request(
method=method,
url=url,
params=params,
data=data,
headers=self._headers,
verify=self._verify_ssl,
timeout=self._timeout
)
break
except (requests.exceptions.ConnectionError,
requests.exceptions.Timeout) as e:
if i < 2:
continue
else:
raise e
if response.status_code == expected_response_code:
return response
else:
raise InfluxDBClientError(response.content, response.status_code)
def write(self, data):
""" Provided as convenience for influxdb v0.9.0, this may change. """
self.request(
url="write",
method='POST',
params=None,
data=data,
expected_response_code=200
)
return True
# Writing Data
#
# Assuming you have a database named foo_production you can write data
# by doing a POST to /db/foo_production/series?u=some_user&p=some_password
# with a JSON body of points.
def write_points(self, data, time_precision='s', *args, **kwargs):
"""
Write to multiple time series names. An example data blob is:
data = [
{
"points": [
[
12
]
],
"name": "cpu_load_short",
"columns": [
"value"
]
}
]
:param data: A list of dicts in InfluxDB 0.8.x data format.
:param time_precision: [Optional, default 's'] Either 's', 'm', 'ms'
or 'u'.
:param batch_size: [Optional] Value to write the points in batches
instead of all at one time. Useful for when doing data dumps from
one database to another or when doing a massive write operation
:type batch_size: int
"""
def list_chunks(l, n):
""" Yield successive n-sized chunks from l.
"""
for i in xrange(0, len(l), n):
yield l[i:i + n]
batch_size = kwargs.get('batch_size')
if batch_size and batch_size > 0:
for item in data:
name = item.get('name')
columns = item.get('columns')
point_list = item.get('points', [])
for batch in list_chunks(point_list, batch_size):
item = [{
"points": batch,
"name": name,
"columns": columns
}]
self._write_points(
data=item,
time_precision=time_precision)
return True
else:
return self._write_points(data=data,
time_precision=time_precision)
def write_points_with_precision(self, data, time_precision='s'):
"""
DEPRECATED. Write to multiple time series names
"""
warnings.warn(
"write_points_with_precision is deprecated, and will be removed "
"in future versions. Please use "
"``InfluxDBClient.write_points(time_precision='..')`` instead.",
FutureWarning)
return self._write_points(data=data, time_precision=time_precision)
def _write_points(self, data, time_precision):
if time_precision not in ['s', 'm', 'ms', 'u']:
raise Exception(
"Invalid time precision is given. (use 's', 'm', 'ms' or 'u')")
if self.use_udp and time_precision != 's':
raise Exception(
"InfluxDB only supports seconds precision for udp writes"
)
url = "db/{0}/series".format(self._database)
params = {
'time_precision': time_precision
}
if self.use_udp:
self.send_packet(data)
else:
self.request(
url=url,
method='POST',
params=params,
data=data,
expected_response_code=200
)
return True
# One Time Deletes
def delete_points(self, name):
"""
Delete an entire series
"""
url = "db/{0}/series/{1}".format(self._database, name)
self.request(
url=url,
method='DELETE',
expected_response_code=204
)
return True
# Regularly Scheduled Deletes
def create_scheduled_delete(self, json_body):
"""
TODO: Create scheduled delete
2013-11-08: This endpoint has not been implemented yet in ver0.0.8,
but it is documented in http://influxdb.org/docs/api/http.html.
See also: src/api/http/api.go:l57
"""
raise NotImplementedError()
# get list of deletes
# curl http://localhost:8086/db/site_dev/scheduled_deletes
#
# remove a regularly scheduled delete
# curl -X DELETE http://localhost:8086/db/site_dev/scheduled_deletes/:id
def get_list_scheduled_delete(self):
"""
TODO: Get list of scheduled deletes
2013-11-08: This endpoint has not been implemented yet in ver0.0.8,
but it is documented in http://influxdb.org/docs/api/http.html.
See also: src/api/http/api.go:l57
"""
raise NotImplementedError()
def remove_scheduled_delete(self, delete_id):
"""
TODO: Remove scheduled delete
2013-11-08: This endpoint has not been implemented yet in ver0.0.8,
but it is documented in http://influxdb.org/docs/api/http.html.
See also: src/api/http/api.go:l57
"""
raise NotImplementedError()
def query(self, query, time_precision='s', chunked=False):
"""
Quering data
:param time_precision: [Optional, default 's'] Either 's', 'm', 'ms'
or 'u'.
:param chunked: [Optional, default=False] True if the data shall be
retrieved in chunks, False otherwise.
"""
return self._query(query, time_precision=time_precision,
chunked=chunked)
# Querying Data
#
# GET db/:name/series. It takes five parameters
def _query(self, query, time_precision='s', chunked=False):
if time_precision not in ['s', 'm', 'ms', 'u']:
raise Exception(
"Invalid time precision is given. (use 's', 'm', 'ms' or 'u')")
if chunked is True:
chunked_param = 'true'
else:
chunked_param = 'false'
# Build the URL of the serie to query
url = "db/{0}/series".format(self._database)
params = {
'q': query,
'time_precision': time_precision,
'chunked': chunked_param
}
response = self.request(
url=url,
method='GET',
params=params,
expected_response_code=200
)
if chunked:
decoded = {}
try:
decoded = chunked_json.loads(response.content.decode())
except UnicodeDecodeError:
decoded = chunked_json.loads(response.content.decode('utf-8'))
finally:
return list(decoded)
else:
return response.json()
# Creating and Dropping Databases
#
# ### create a database
# curl -X POST http://localhost:8086/db -d '{"name": "site_development"}'
#
# ### drop a database
# curl -X DELETE http://localhost:8086/db/site_development
def create_database(self, database):
"""
create_database()
Create a database on the InfluxDB server.
:param database: the name of the database to create
:type database: string
:rtype: boolean
"""
url = "db"
data = {'name': database}
self.request(
url=url,
method='POST',
data=data,
expected_response_code=201
)
return True
def delete_database(self, database):
"""
delete_database()
Drop a database on the InfluxDB server.
:param database: the name of the database to delete
:type database: string
:rtype: boolean
"""
url = "db/{0}".format(database)
self.request(
url=url,
method='DELETE',
expected_response_code=204
)
return True
# ### get list of databases
# curl -X GET http://localhost:8086/db
def get_list_database(self):
"""
Get the list of databases
"""
url = "db"
response = self.request(
url=url,
method='GET',
expected_response_code=200
)
return response.json()
def get_database_list(self):
"""
DEPRECATED. Get the list of databases
"""
warnings.warn(
"get_database_list is deprecated, and will be removed "
"in future versions. Please use "
"``InfluxDBClient.get_list_database`` instead.",
FutureWarning)
return self.get_list_database()
def delete_series(self, series):
"""
delete_series()
Drop a series on the InfluxDB server.
:param series: the name of the series to delete
:type series: string
:rtype: boolean
"""
url = "db/{0}/series/{1}".format(
self._database,
series
)
self.request(
url=url,
method='DELETE',
expected_response_code=204
)
return True
def get_list_series(self):
"""
Get a list of all time series in a database
"""
response = self._query('list series')
series_list = []
for series in response[0]['points']:
series_list.append(series[1])
return series_list
def get_list_continuous_queries(self):
"""
Get a list of continuous queries
"""
response = self._query('list continuous queries')
queries_list = []
for query in response[0]['points']:
queries_list.append(query[2])
return queries_list
# Security
# get list of cluster admins
# curl http://localhost:8086/cluster_admins?u=root&p=root
# add cluster admin
# curl -X POST http://localhost:8086/cluster_admins?u=root&p=root \
# -d '{"name": "paul", "password": "i write teh docz"}'
# update cluster admin password
# curl -X POST http://localhost:8086/cluster_admins/paul?u=root&p=root \
# -d '{"password": "new pass"}'
# delete cluster admin
# curl -X DELETE http://localhost:8086/cluster_admins/paul?u=root&p=root
# Database admins, with a database name of site_dev
# get list of database admins
# curl http://localhost:8086/db/site_dev/admins?u=root&p=root
# add database admin
# curl -X POST http://localhost:8086/db/site_dev/admins?u=root&p=root \
# -d '{"name": "paul", "password": "i write teh docz"}'
# update database admin password
# curl -X POST http://localhost:8086/db/site_dev/admins/paul?u=root&p=root\
# -d '{"password": "new pass"}'
# delete database admin
# curl -X DELETE \
# http://localhost:8086/db/site_dev/admins/paul?u=root&p=root
def get_list_cluster_admins(self):
"""
Get list of cluster admins
"""
response = self.request(
url="cluster_admins",
method='GET',
expected_response_code=200
)
return response.json()
def add_cluster_admin(self, new_username, new_password):
"""
Add cluster admin
"""
data = {
'name': new_username,
'password': new_password
}
self.request(
url="cluster_admins",
method='POST',
data=data,
expected_response_code=200
)
return True
def update_cluster_admin_password(self, username, new_password):
"""
Update cluster admin password
"""
url = "cluster_admins/{0}".format(username)
data = {
'password': new_password
}
self.request(
url=url,
method='POST',
data=data,
expected_response_code=200
)
return True
def delete_cluster_admin(self, username):
"""
Delete cluster admin
"""
url = "cluster_admins/{0}".format(username)
self.request(
url=url,
method='DELETE',
expected_response_code=200
)
return True
def set_database_admin(self, username):
"""
Set user as database admin
"""
return self.alter_database_admin(username, True)
def unset_database_admin(self, username):
"""
Unset user as database admin
"""
return self.alter_database_admin(username, False)
def alter_database_admin(self, username, is_admin):
url = "db/{0}/users/{1}".format(self._database, username)
data = {'admin': is_admin}
self.request(
url=url,
method='POST',
data=data,
expected_response_code=200
)
return True
def get_list_database_admins(self):
"""
TODO: Get list of database admins
2013-11-08: This endpoint has not been implemented yet in ver0.0.8,
but it is documented in http://influxdb.org/docs/api/http.html.
See also: src/api/http/api.go:l57
"""
raise NotImplementedError()
def add_database_admin(self, new_username, new_password):
"""
TODO: Add cluster admin
2013-11-08: This endpoint has not been implemented yet in ver0.0.8,
but it is documented in http://influxdb.org/docs/api/http.html.
See also: src/api/http/api.go:l57
"""
raise NotImplementedError()
def update_database_admin_password(self, username, new_password):
"""
TODO: Update database admin password
2013-11-08: This endpoint has not been implemented yet in ver0.0.8,
but it is documented in http://influxdb.org/docs/api/http.html.
See also: src/api/http/api.go:l57
"""
raise NotImplementedError()
def delete_database_admin(self, username):
"""
TODO: Delete database admin
2013-11-08: This endpoint has not been implemented yet in ver0.0.8,
but it is documented in http://influxdb.org/docs/api/http.html.
See also: src/api/http/api.go:l57
"""
raise NotImplementedError()
###
# Limiting User Access
# Database users
# get list of database users
# curl http://localhost:8086/db/site_dev/users?u=root&p=root
# add database user
# curl -X POST http://localhost:8086/db/site_dev/users?u=root&p=root \
# -d '{"name": "paul", "password": "i write teh docz"}'
# update database user password
# curl -X POST http://localhost:8086/db/site_dev/users/paul?u=root&p=root \
# -d '{"password": "new pass"}'
# delete database user
# curl -X DELETE http://localhost:8086/db/site_dev/users/paul?u=root&p=root
def get_database_users(self):
"""
Get list of database users
"""
url = "db/{0}/users".format(self._database)
response = self.request(
url=url,
method='GET',
expected_response_code=200
)
return response.json()
def add_database_user(self, new_username, new_password, permissions=None):
"""
Add database user
:param permissions: A ``(readFrom, writeTo)`` tuple
"""
url = "db/{0}/users".format(self._database)
data = {
'name': new_username,
'password': new_password
}
if permissions:
try:
data['readFrom'], data['writeTo'] = permissions
except (ValueError, TypeError):
raise TypeError(
"'permissions' must be (readFrom, writeTo) tuple"
)
self.request(
url=url,
method='POST',
data=data,
expected_response_code=200
)
return True
def update_database_user_password(self, username, new_password):
"""
Update password
"""
return self.alter_database_user(username, new_password)
def alter_database_user(self, username, password=None, permissions=None):
"""
Alters a database user and/or their permissions.
:param permissions: A ``(readFrom, writeTo)`` tuple
:raise TypeError: if permissions cannot be read.
:raise ValueError: if neither password nor permissions provided.
"""
url = "db/{0}/users/{1}".format(self._database, username)
if not password and not permissions:
raise ValueError("Nothing to alter for user {}.".format(username))
data = {}
if password:
data['password'] = password
if permissions:
try:
data['readFrom'], data['writeTo'] = permissions
except (ValueError, TypeError):
raise TypeError(
"'permissions' must be (readFrom, writeTo) tuple"
)
self.request(
url=url,
method='POST',
data=data,
expected_response_code=200
)
if username == self._username:
self._password = password
return True
def delete_database_user(self, username):
"""
Delete database user
"""
url = "db/{0}/users/{1}".format(self._database, username)
self.request(
url=url,
method='DELETE',
expected_response_code=200
)
return True
# update the user by POSTing to db/site_dev/users/paul
def update_permission(self, username, json_body):
"""
TODO: Update read/write permission
2013-11-08: This endpoint has not been implemented yet in ver0.0.8,
but it is documented in http://influxdb.org/docs/api/http.html.
See also: src/api/http/api.go:l57
"""
raise NotImplementedError()
def send_packet(self, packet):
data = json.dumps(packet)
byte = data.encode('utf-8')
self.udp_socket.sendto(byte, (self._host, self.udp_port))
PK EpG;p5 5 ( influxdb-2.8.0.dist-info/DESCRIPTION.rst
InfluxDB-Python is a client for interacting with InfluxDB_.
.. image:: https://travis-ci.org/influxdb/influxdb-python.svg?branch=master
:target: https://travis-ci.org/influxdb/influxdb-python
.. image:: https://readthedocs.org/projects/influxdb-python/badge/?version=latest&style
:target: http://influxdb-python.readthedocs.org/
:alt: Documentation Status
.. image:: https://img.shields.io/coveralls/influxdb/influxdb-python.svg
:target: https://coveralls.io/r/influxdb/influxdb-python
:alt: Coverage
.. image:: https://pypip.in/download/influxdb/badge.svg
:target: https://pypi.python.org/pypi//influxdb/
:alt: Downloads
.. image:: https://pypip.in/version/influxdb/badge.svg
:target: https://pypi.python.org/pypi/influxdb/
:alt: Latest Version
.. image:: https://pypip.in/py_versions/influxdb/badge.svg
:target: https://pypi.python.org/pypi/influxdb/
:alt: Supported Python versions
.. image:: https://pypip.in/license/influxdb/badge.svg
:target: https://pypi.python.org/pypi/influxdb/
:alt: License
.. _readme-about:
InfluxDB is an open-source distributed time series database, find more about InfluxDB_ at http://influxdb.com/
.. _installation:
InfluxDB v0.8.X users
=====================
InfluxDB 0.9 was released and it is the new recommended version. However, InfluxDB 0.8.x users may still use the legacy client by using ``from influxdb.influxdb08 import InfluxDBClient`` instead.
Installation
============
Install, upgrade and uninstall InfluxDB-Python with these commands::
$ pip install influxdb
$ pip install --upgrade influxdb
$ pip uninstall influxdb
On Debian/Ubuntu, you can install it with this command::
$ sudo apt-get install python-influxdb
Dependencies
============
The InfluxDB-Python distribution is supported and tested on Python 2.7, 3.2, 3.3, 3.4, PyPy and PyPy3.
Main dependency is:
- Requests: HTTP library for human beings (http://docs.python-requests.org/)
Additional dependencies are:
- pandas: for writing from and reading to DataFrames (http://pandas.pydata.org/)
- Sphinx: Tool to create and manage the documentation (http://sphinx-doc.org/)
- Nose: to auto-discover tests (http://nose.readthedocs.org/en/latest/)
- Mock: to mock tests (https://pypi.python.org/pypi/mock)
Documentation
=============
InfluxDB-Python documentation is available at http://influxdb-python.readthedocs.org
You will need Sphinx_ installed to generate the documentation.
The documentation can be generated by running::
$ tox -e docs
Generated documentation can be found in the *docs/build/html/* directory.
Examples
========
Here's a basic example (for more see the examples directory)::
$ python
>>> from influxdb import InfluxDBClient
>>> json_body = [
{
"measurement": "cpu_load_short",
"tags": {
"host": "server01",
"region": "us-west"
},
"time": "2009-11-10T23:00:00Z",
"fields": {
"value": 0.64
}
}
]
>>> client = InfluxDBClient('localhost', 8086, 'root', 'root', 'example')
>>> client.create_database('example')
>>> client.write_points(json_body)
>>> result = client.query('select value from cpu_load_short;')
>>> print("Result: {0}".format(result))
If you want to connect to a cluster, you could initialize a ``InfluxDBClusterClient``::
$ python
>>> from influxdb import InfluxDBClusterClient
>>> cc = InfluxDBClusterClient(hosts = [('192.168.0.1', 8086),
('192.168.0.2', 8086),
('192.168.0.3', 8086)],
username='root',
password='root',
database='example')
``InfluxDBClusterClient`` has the same methods as ``InfluxDBClient``, it basically is a proxy to multiple InfluxDBClients.
Testing
=======
Make sure you have tox by running the following::
$ pip install tox
To test influxdb-python with multiple version of Python, you can use Tox_::
$ tox
Support
=======
For issues with, questions about, or feedback for InfluxDB_, please look into
our community page: http://influxdb.com/community/.
Development
===========
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
TODO
====
The TODO/Roadmap can be found in Github bug tracker: https://github.com/influxdb/influxdb-python/issues/109
Source code
===========
The source code is currently available on Github: https://github.com/influxdb/influxdb-python
.. _InfluxDB: http://influxdb.com/
.. _Sphinx: http://sphinx.pocoo.org/
.. _Tox: https://tox.readthedocs.org
PK EpGoz & influxdb-2.8.0.dist-info/metadata.json{"classifiers": ["Development Status :: 3 - Alpha", "Intended Audience :: Developers", "License :: OSI Approved :: MIT License", "Operating System :: OS Independent", "Programming Language :: Python", "Programming Language :: Python :: 2.7", "Programming Language :: Python :: 3", "Programming Language :: Python :: 3.3", "Topic :: Software Development :: Libraries", "Topic :: Software Development :: Libraries :: Python Modules"], "extensions": {"python.details": {"document_names": {"description": "DESCRIPTION.rst"}, "project_urls": {"Home": "https://github.com/influxdb/influxdb-python"}}}, "extras": ["test"], "generator": "bdist_wheel (0.24.0)", "license": "MIT License", "metadata_version": "2.0", "name": "influxdb", "run_requires": [{"extra": "test", "requires": ["nose", "nose-cov", "mock", "requests-mock"]}, {"requires": ["python-dateutil (>=2.0.0)", "pytz", "requests (>=1.0.3)", "six (==1.9.0)"]}], "summary": "InfluxDB client", "test_requires": [{"requires": ["nose", "nose-cov", "mock", "requests-mock"]}], "version": "2.8.0"}PK EpG-C# & influxdb-2.8.0.dist-info/top_level.txtinfluxdb
PK EpG3on n influxdb-2.8.0.dist-info/WHEELWheel-Version: 1.0
Generator: bdist_wheel (0.24.0)
Root-Is-Purelib: true
Tag: py2-none-any
Tag: py3-none-any
PK EpGۙT2 2 ! influxdb-2.8.0.dist-info/METADATAMetadata-Version: 2.0
Name: influxdb
Version: 2.8.0
Summary: InfluxDB client
Home-page: https://github.com/influxdb/influxdb-python
Author: UNKNOWN
Author-email: UNKNOWN
License: MIT License
Platform: UNKNOWN
Classifier: Development Status :: 3 - Alpha
Classifier: Intended Audience :: Developers
Classifier: License :: OSI Approved :: MIT License
Classifier: Operating System :: OS Independent
Classifier: Programming Language :: Python
Classifier: Programming Language :: Python :: 2.7
Classifier: Programming Language :: Python :: 3
Classifier: Programming Language :: Python :: 3.3
Classifier: Topic :: Software Development :: Libraries
Classifier: Topic :: Software Development :: Libraries :: Python Modules
Requires-Dist: python-dateutil (>=2.0.0)
Requires-Dist: pytz
Requires-Dist: requests (>=1.0.3)
Requires-Dist: six (==1.9.0)
Provides-Extra: test
Requires-Dist: nose; extra == 'test'
Requires-Dist: nose-cov; extra == 'test'
Requires-Dist: mock; extra == 'test'
Requires-Dist: requests-mock; extra == 'test'
InfluxDB-Python is a client for interacting with InfluxDB_.
.. image:: https://travis-ci.org/influxdb/influxdb-python.svg?branch=master
:target: https://travis-ci.org/influxdb/influxdb-python
.. image:: https://readthedocs.org/projects/influxdb-python/badge/?version=latest&style
:target: http://influxdb-python.readthedocs.org/
:alt: Documentation Status
.. image:: https://img.shields.io/coveralls/influxdb/influxdb-python.svg
:target: https://coveralls.io/r/influxdb/influxdb-python
:alt: Coverage
.. image:: https://pypip.in/download/influxdb/badge.svg
:target: https://pypi.python.org/pypi//influxdb/
:alt: Downloads
.. image:: https://pypip.in/version/influxdb/badge.svg
:target: https://pypi.python.org/pypi/influxdb/
:alt: Latest Version
.. image:: https://pypip.in/py_versions/influxdb/badge.svg
:target: https://pypi.python.org/pypi/influxdb/
:alt: Supported Python versions
.. image:: https://pypip.in/license/influxdb/badge.svg
:target: https://pypi.python.org/pypi/influxdb/
:alt: License
.. _readme-about:
InfluxDB is an open-source distributed time series database, find more about InfluxDB_ at http://influxdb.com/
.. _installation:
InfluxDB v0.8.X users
=====================
InfluxDB 0.9 was released and it is the new recommended version. However, InfluxDB 0.8.x users may still use the legacy client by using ``from influxdb.influxdb08 import InfluxDBClient`` instead.
Installation
============
Install, upgrade and uninstall InfluxDB-Python with these commands::
$ pip install influxdb
$ pip install --upgrade influxdb
$ pip uninstall influxdb
On Debian/Ubuntu, you can install it with this command::
$ sudo apt-get install python-influxdb
Dependencies
============
The InfluxDB-Python distribution is supported and tested on Python 2.7, 3.2, 3.3, 3.4, PyPy and PyPy3.
Main dependency is:
- Requests: HTTP library for human beings (http://docs.python-requests.org/)
Additional dependencies are:
- pandas: for writing from and reading to DataFrames (http://pandas.pydata.org/)
- Sphinx: Tool to create and manage the documentation (http://sphinx-doc.org/)
- Nose: to auto-discover tests (http://nose.readthedocs.org/en/latest/)
- Mock: to mock tests (https://pypi.python.org/pypi/mock)
Documentation
=============
InfluxDB-Python documentation is available at http://influxdb-python.readthedocs.org
You will need Sphinx_ installed to generate the documentation.
The documentation can be generated by running::
$ tox -e docs
Generated documentation can be found in the *docs/build/html/* directory.
Examples
========
Here's a basic example (for more see the examples directory)::
$ python
>>> from influxdb import InfluxDBClient
>>> json_body = [
{
"measurement": "cpu_load_short",
"tags": {
"host": "server01",
"region": "us-west"
},
"time": "2009-11-10T23:00:00Z",
"fields": {
"value": 0.64
}
}
]
>>> client = InfluxDBClient('localhost', 8086, 'root', 'root', 'example')
>>> client.create_database('example')
>>> client.write_points(json_body)
>>> result = client.query('select value from cpu_load_short;')
>>> print("Result: {0}".format(result))
If you want to connect to a cluster, you could initialize a ``InfluxDBClusterClient``::
$ python
>>> from influxdb import InfluxDBClusterClient
>>> cc = InfluxDBClusterClient(hosts = [('192.168.0.1', 8086),
('192.168.0.2', 8086),
('192.168.0.3', 8086)],
username='root',
password='root',
database='example')
``InfluxDBClusterClient`` has the same methods as ``InfluxDBClient``, it basically is a proxy to multiple InfluxDBClients.
Testing
=======
Make sure you have tox by running the following::
$ pip install tox
To test influxdb-python with multiple version of Python, you can use Tox_::
$ tox
Support
=======
For issues with, questions about, or feedback for InfluxDB_, please look into
our community page: http://influxdb.com/community/.
Development
===========
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
TODO
====
The TODO/Roadmap can be found in Github bug tracker: https://github.com/influxdb/influxdb-python/issues/109
Source code
===========
The source code is currently available on Github: https://github.com/influxdb/influxdb-python
.. _InfluxDB: http://influxdb.com/
.. _Sphinx: http://sphinx.pocoo.org/
.. _Tox: https://tox.readthedocs.org
PK EpG{{=y y influxdb-2.8.0.dist-info/RECORDinfluxdb/exceptions.py,sha256=sikc5pR3T9WnB2YcbTXzlB3gGq3aT3LfeoPii44RgkA,681
influxdb/__init__.py,sha256=wkO005crAFwwzxiRUCBUB4vqCbGo8wjKTAC_nNGQII0,314
influxdb/line_protocol.py,sha256=ynK2xj1W5x2TxB8THNBliL0cIy4eqsm-pljBrc5HqKM,3765
influxdb/resultset.py,sha256=Fj34cVLFbWLHnrLMMOCXQmN2aXJhBP0-zu6WWB_ZUn4,5983
influxdb/_dataframe_client.py,sha256=pX0CHAelbktH4yWv5ULRcpsbDfnq0wcfOeTC1X_HaZ4,5849
influxdb/dataframe_client.py,sha256=vpD4w0f_i5gPjqk9-GXr6WBx0drFprBOc-_l6UXfrD8,472
influxdb/helper.py,sha256=gkQKsTtW873ZSDwu9BOL4bmjA_v1YXEDANk7odaAYvU,5560
influxdb/chunked_json.py,sha256=vNGDwD1U1k0BA5Qg4iYXK5IFtzG-8dbQSMJcroeYYcE,454
influxdb/client.py,sha256=6VxVxrl2ObHgyZucL7L9tlJwHjHIfO0fOLLIY19nJxM,29535
influxdb/tests/__init__.py,sha256=AtgMGlmuwh-NX3JPQLo5-hPPAyXEFQ0gb-NFcOH-iNw,396
influxdb/tests/misc.py,sha256=ExU_J7OJINfEvphJjsvMFnp1nP1bozriTn9KEdcAXZ0,693
influxdb/tests/chunked_json_test.py,sha256=Uk0qxVrU-4fa3kbAIrj4Ur_dyBtI-FOnWOooHeZ8Uqk,1482
influxdb/tests/resultset_test.py,sha256=qEWeiSq7pzqueCRSXKGssGeoiTWYawBm496RCnfA6zQ,5553
influxdb/tests/test_line_protocol.py,sha256=xvXu-DEF0K_omYQQ-Q_ILHVHVMiOv0tFj-hvxQRWy2A,1266
influxdb/tests/dataframe_client_test.py,sha256=O2YW4S2tA-8l2hZLNJTd2ROhV7bWDxdrcKaPUyUw768,13335
influxdb/tests/client_test.py,sha256=JPHneGr71DG1fLNPJLI4i-mY32boYY-uSuy0lQJVq80,30602
influxdb/tests/helper_test.py,sha256=xQ5KkyeOg_Llc6lDZ9oSM-nfyQt8daBY6Q8rrYmvFGw,9584
influxdb/tests/server_tests/__init__.py,sha256=47DEQpj8HBSa-_TImW-5JCeuQeRkm5NMpJWZG3hSuFU,0
influxdb/tests/server_tests/influxdb_instance.py,sha256=ccbCXq7acrmvCQn3UtuI_ZTOUQN3uT0rVxguLcwhuU8,5519
influxdb/tests/server_tests/base.py,sha256=XD0srjpmA69Kio_ya3eYmxDEvGURc6QbJ3BzLyAeedo,2031
influxdb/tests/server_tests/client_test_with_server.py,sha256=mK6sD6WmSm7AlQ-ivhoFkeEhnOW3IyhUCZWA-ylV6Wc,23487
influxdb/tests/influxdb08/__init__.py,sha256=iwhKnzeBJLKxpRVjvzwiRE63_zNpIBfaKLITauVph-0,24
influxdb/tests/influxdb08/dataframe_client_test.py,sha256=9SvmJvK9_sISghf0LkwB4WZu3Vi0Imx8CZM0RPO47v4,12409
influxdb/tests/influxdb08/client_test.py,sha256=jJE19cJzJEqh_2Gnmgt1-w4F7Xw9L_ULXHnxqI_ekE4,28011
influxdb/tests/influxdb08/helper_test.py,sha256=VFQTFtVxmMuTzd1Y1TbC0D19oovgSBWUo0ADwEK5fts,8175
influxdb/influxdb08/__init__.py,sha256=jG3dRrqE_0ww5oarXV0pjEpUtN1v6vSp5NgvjgJ8SCU,219
influxdb/influxdb08/dataframe_client.py,sha256=olVccxdL81vOdpynFEd-zeh3ApdZjhRJ2kO6Gqlk9Y0,6828
influxdb/influxdb08/helper.py,sha256=WGbR0xzkttYcZTngfIcqnnkMMgSBa-r9cyWywPRyALc,5308
influxdb/influxdb08/chunked_json.py,sha256=vNGDwD1U1k0BA5Qg4iYXK5IFtzG-8dbQSMJcroeYYcE,454
influxdb/influxdb08/client.py,sha256=vNaohc9_2VEBkuVRyBYJyO9ktb5csMnvISPF-GIB1Y8,26237
influxdb-2.8.0.dist-info/WHEEL,sha256=AvR0WeTpDaxT645bl5FQxUK6NPsTls2ttpcGJg3j1Xg,110
influxdb-2.8.0.dist-info/metadata.json,sha256=UmmwaYnEuhTIWrz0RLNVWNNVkyGa8G2gok_A1wvPvng,1043
influxdb-2.8.0.dist-info/RECORD,,
influxdb-2.8.0.dist-info/DESCRIPTION.rst,sha256=kRDiO6_CfpWWCD2cPzkwYXnOQvraCIf6AqQjuD-pfAA,4917
influxdb-2.8.0.dist-info/top_level.txt,sha256=5OqJcANRKMlrogGgl2mHmihApnH6UADfcpprDdmWj88,9
influxdb-2.8.0.dist-info/METADATA,sha256=VKeSzZPQmC0zmoQUtCM8cfNaeXtOhzDOSvKS5N7asj4,5938
PK