Documentation

InfluxDB 2.0 client features

Installation

Note

It is recommended to useciso8601 with client for parsing dates.ciso8601 is much faster than built-in Python datetime. Since it's written as aC module the best way is build it from sources:

Windows:

You have to installVisual C++ Build Tools 2015 to buildciso8601 bypip.

conda:

Install from sources:conda install -c conda-forge/label/cf202003 ciso8601.

pip install

pip install'influxdb-client'

importinfluxdb_client

Setuptools

python setup.py install --user

Getting Started

frominfluxdb_clientimportInfluxDBClient,Pointfrominfluxdb_client.client.write_apiimportSYNCHRONOUSbucket="my-bucket"client=InfluxDBClient(url="http://localhost:8086",token="my-token",org="my-org")write_api=client.write_api(write_options=SYNCHRONOUS)query_api=client.query_api()p=Point("my_measurement").tag("location","Prague").field("temperature",25.3)write_api.write(bucket=bucket,record=p)## using Table structuretables=query_api.query('from(bucket:"my-bucket") |> range(start: -10m)')fortableintables:print(table)forrowintable.records:print (row.values)## using csv librarycsv_result=query_api.query_csv('from(bucket:"my-bucket") |> range(start: -10m)')val_count=0forrowincsv_result:forcellinrow:val_count+=1

Client configuration

Via File

self.client=InfluxDBClient.from_config_file("config.ini")

Via Environment Properties

self.client=InfluxDBClient.from_env_properties()

Profile query

q='''    from(bucket: stringParam)      |> range(start: -5m, stop: now())      |> filter(fn: (r) => r._measurement == "mem")      |> filter(fn: (r) => r._field == "available" or r._field == "free" or r._field == "used")      |> aggregateWindow(every: 1m, fn: mean)      |> pivot(rowKey:["_time"], columnKey: ["_field"], valueColumn: "_value")'''p= {"stringParam":"my-bucket",}query_api=client.query_api(query_options=QueryOptions(profilers=["query","operator"]))csv_result=query_api.query(query=q,params=p)

classProfilersCallback(object):def__init__(self):self.records= []def__call__(self,flux_record):self.records.append(flux_record.values)callback=ProfilersCallback()query_api=client.query_api(query_options=QueryOptions(profilers=["query","operator"],profiler_callback=callback))tables=query_api.query('from(bucket:"my-bucket") |> range(start: -10m)')forprofilerincallback.records:print(f'Custom processing of profiler result:{profiler}')

How to use

Writes

Warning

TheWriteApi in batching mode (default mode) is suppose to run as a singleton.To flush all your data you should wrap the execution usingwith client.write_api(...) as write_api: statementor callwrite_api.close() at the end of your script.

The data could be written as

Batching

fromdatetimeimportdatetime,timedeltaimportpandasaspdimportreactivexasrxfromreactiveximportoperatorsasopsfrominfluxdb_clientimportInfluxDBClient,Point,WriteOptionswithInfluxDBClient(url="http://localhost:8086",token="my-token",org="my-org")as_client:with_client.write_api(write_options=WriteOptions(batch_size=500,flush_interval=10_000,jitter_interval=2_000,retry_interval=5_000,max_retries=5,max_retry_delay=30_000,exponential_base=2))as_write_client:"""        Write Line Protocol formatted as string        """_write_client.write("my-bucket","my-org","h2o_feet,location=coyote_creek water_level=1.0 1")_write_client.write("my-bucket","my-org", ["h2o_feet,location=coyote_creek water_level=2.0 2","h2o_feet,location=coyote_creek water_level=3.0 3"])"""        Write Line Protocol formatted as byte array        """_write_client.write("my-bucket","my-org","h2o_feet,location=coyote_creek water_level=1.0 1".encode())_write_client.write("my-bucket","my-org", ["h2o_feet,location=coyote_creek water_level=2.0 2".encode(),"h2o_feet,location=coyote_creek water_level=3.0 3".encode()])"""        Write Dictionary-style object        """_write_client.write("my-bucket","my-org", {"measurement":"h2o_feet","tags": {"location":"coyote_creek"},"fields": {"water_level":1.0},"time":1})_write_client.write("my-bucket","my-org", [{"measurement":"h2o_feet","tags": {"location":"coyote_creek"},"fields": {"water_level":2.0},"time":2},                                                    {"measurement":"h2o_feet","tags": {"location":"coyote_creek"},"fields": {"water_level":3.0},"time":3}])"""        Write Data Point        """_write_client.write("my-bucket","my-org",Point("h2o_feet").tag("location","coyote_creek").field("water_level",4.0).time(4))_write_client.write("my-bucket","my-org",                            [Point("h2o_feet").tag("location","coyote_creek").field("water_level",5.0).time(5),Point("h2o_feet").tag("location","coyote_creek").field("water_level",6.0).time(6)])"""        Write Observable stream        """_data=rx \            .range(7,11) \            .pipe(ops.map(lambdai:"h2o_feet,location=coyote_creek water_level={0}.0 {0}".format(i)))_write_client.write("my-bucket","my-org",_data)"""        Write Pandas DataFrame        """_now=datetime.utcnow()_data_frame=pd.DataFrame(data=[["coyote_creek",1.0], ["coyote_creek",2.0]],index=[_now,_now+timedelta(hours=1)],columns=["location","water_level"])_write_client.write("my-bucket","my-org",record=_data_frame,data_frame_measurement_name='h2o_feet',data_frame_tag_columns=['location'])

Default Tags

Via API

point_settings=PointSettings()point_settings.add_default_tag("id","132-987-655")point_settings.add_default_tag("customer","California Miner")point_settings.add_default_tag("data_center","${env.data_center}")self.write_client=self.client.write_api(write_options=SYNCHRONOUS,point_settings=point_settings)

self.write_client=self.client.write_api(write_options=SYNCHRONOUS,point_settings=PointSettings(**{"id":"132-987-655","customer":"California Miner"}))

Via Configuration file

self.client=InfluxDBClient.from_config_file("config.ini")

Via Environment Properties

self.client=InfluxDBClient.from_env_properties()

Synchronous client

frominfluxdb_clientimportInfluxDBClient,Pointfrominfluxdb_client .client.write_apiimportSYNCHRONOUSclient=InfluxDBClient(url="http://localhost:8086",token="my-token",org="my-org")write_api=client.write_api(write_options=SYNCHRONOUS)_point1=Point("my_measurement").tag("location","Prague").field("temperature",25.3)_point2=Point("my_measurement").tag("location","New York").field("temperature",24.3)write_api.write(bucket="my-bucket",record=[_point1,_point2])client.close()

Queries

frominfluxdb_clientimportInfluxDBClient,Point,Dialectfrominfluxdb_client.client.write_apiimportSYNCHRONOUSclient=InfluxDBClient(url="http://localhost:8086",token="my-token",org="my-org")write_api=client.write_api(write_options=SYNCHRONOUS)query_api=client.query_api()"""Prepare data"""_point1=Point("my_measurement").tag("location","Prague").field("temperature",25.3)_point2=Point("my_measurement").tag("location","New York").field("temperature",24.3)write_api.write(bucket="my-bucket",record=[_point1,_point2])"""Query: using Table structure"""tables=query_api.query('from(bucket:"my-bucket") |> range(start: -10m)')fortableintables:print(table)forrecordintable.records:print(record.values)print()print()"""Query: using Bind parameters"""p= {"_start":datetime.timedelta(hours=-1),"_location":"Prague","_desc":True,"_floatParam":25.1,"_every":datetime.timedelta(minutes=5)     }tables=query_api.query('''    from(bucket:"my-bucket") |> range(start: _start)        |> filter(fn: (r) => r["_measurement"] == "my_measurement")        |> filter(fn: (r) => r["_field"] == "temperature")        |> filter(fn: (r) => r["location"] == _location and r["_value"] > _floatParam)        |> aggregateWindow(every: _every, fn: mean, createEmpty: true)        |> sort(columns: ["_time"], desc: _desc)''',params=p)fortableintables:print(table)forrecordintable.records:print(str(record["_time"])+" - "+record["location"]+": "+str(record["_value"]))print()print()"""Query: using Stream"""records=query_api.query_stream('from(bucket:"my-bucket") |> range(start: -10m)')forrecordinrecords:print(f'Temperature in{record["location"]} is{record["_value"]}')"""Interrupt a stream after retrieve a required data"""large_stream=query_api.query_stream('from(bucket:"my-bucket") |> range(start: -100d)')forrecordinlarge_stream:ifrecord["location"]=="New York":print(f'New York temperature:{record["_value"]}')breaklarge_stream.close()print()print()"""Query: using csv library"""csv_result=query_api.query_csv('from(bucket:"my-bucket") |> range(start: -10m)',dialect=Dialect(header=False,delimiter=",",comment_prefix="#",annotations=[],date_time_format="RFC3339"))forcsv_lineincsv_result:ifnotlen(csv_line)==0:print(f'Temperature in{csv_line[9]} is{csv_line[6]}')"""Close client"""client.close()

Pandas DataFrame

Note

For DataFrame querying you should install Pandas dependency viapip install 'influxdb-client[extra]'.

Note

Note that if a query returns more then one table then the client generates aDataFrame for each of them.

frominfluxdb_clientimportInfluxDBClient,Point,Dialectfrominfluxdb_client.client.write_apiimportSYNCHRONOUSclient=InfluxDBClient(url="http://localhost:8086",token="my-token",org="my-org")write_api=client.write_api(write_options=SYNCHRONOUS)query_api=client.query_api()"""Prepare data"""_point1=Point("my_measurement").tag("location","Prague").field("temperature",25.3)_point2=Point("my_measurement").tag("location","New York").field("temperature",24.3)write_api.write(bucket="my-bucket",record=[_point1,_point2])"""Query: using Pandas DataFrame"""data_frame=query_api.query_data_frame('from(bucket:"my-bucket") ''|> range(start: -10m) ''|> pivot(rowKey:["_time"], columnKey: ["_field"], valueColumn: "_value") ''|> keep(columns: ["location", "temperature"])')print(data_frame.to_string())"""Close client"""client.close()

Examples

How to efficiently import large dataset

"""Import VIX - CBOE Volatility Index - from "vix-daily.csv" file into InfluxDB 2.0https://datahub.io/core/finance-vix#data"""fromcollectionsimportOrderedDictfromcsvimportDictReaderimportreactivexasrxfromreactiveximportoperatorsasopsfrominfluxdb_clientimportInfluxDBClient,Point,WriteOptionsdefparse_row(row:OrderedDict):"""Parse row of CSV file into Point with structure:        financial-analysis,type=ily close=18.47,high=19.82,low=18.28,open=19.82 1198195200000000000    CSV format:        Date,VIX Open,VIX High,VIX Low,VIX Close\n        2004-01-02,17.96,18.68,17.54,18.22\n        2004-01-05,18.45,18.49,17.44,17.49\n        2004-01-06,17.66,17.67,16.19,16.73\n        2004-01-07,16.72,16.75,15.5,15.5\n        2004-01-08,15.42,15.68,15.32,15.61\n        2004-01-09,16.15,16.88,15.57,16.75\n        ...    :param row: the row of CSV file    :return: Parsed csv row to [Point]    """"""     For better performance is sometimes useful directly create a LineProtocol to avoid unnecessary escaping overhead:     """# from datetime import timezone# import ciso8601# from influxdb_client.client.write.point import EPOCH## time = (ciso8601.parse_datetime(row["Date"]).replace(tzinfo=timezone.utc) - EPOCH).total_seconds() * 1e9# return f"financial-analysis,type=vix-daily" \#        f" close={float(row['VIX Close'])},high={float(row['VIX High'])},low={float(row['VIX Low'])},open={float(row['VIX Open'])} " \#        f" {int(time)}"returnPoint("financial-analysis") \        .tag("type","vix-daily") \        .field("open",float(row['VIX Open'])) \        .field("high",float(row['VIX High'])) \        .field("low",float(row['VIX Low'])) \        .field("close",float(row['VIX Close'])) \        .time(row['Date'])"""Converts vix-daily.csv into sequence of datad point"""data=rx \    .from_iterable(DictReader(open('vix-daily.csv','r'))) \    .pipe(ops.map(lambdarow:parse_row(row)))client=InfluxDBClient(url="http://localhost:8086",token="my-token",org="my-org",debug=True)"""Create client that writes data in batches with 50_000 items."""write_api=client.write_api(write_options=WriteOptions(batch_size=50_000,flush_interval=10_000))"""Write data into InfluxDB"""write_api.write(bucket="my-bucket",record=data)write_api.close()"""Querying max value of CBOE Volatility Index"""query='from(bucket:"my-bucket")' \' |> range(start: 0, stop: now())' \' |> filter(fn: (r) => r._measurement == "financial-analysis")' \' |> max()'result=client.query_api().query(query=query)"""Processing results"""print()print("=== results ===")print()fortableinresult:forrecordintable.records:print('max {0:5} = {1}'.format(record.get_field(),record.get_value()))"""Close client"""client.close()

Efficiency write data from IOT sensor

"""Efficiency write data from IOT sensor - write changed temperature every minute"""importatexitimportplatformfromdatetimeimporttimedeltaimportpsutilaspsutilimportreactivexasrxfromreactiveximportoperatorsasopsfrominfluxdb_clientimportInfluxDBClient,WriteApi,WriteOptionsdefon_exit(db_client:InfluxDBClient,write_api:WriteApi):"""Close clients after terminate a script.    :param db_client: InfluxDB client    :param write_api: WriteApi    :return: nothing    """write_api.close()db_client.close()defsensor_temperature():"""Read a CPU temperature. The [psutil] doesn't support MacOS so we use [sysctl].    :return: actual CPU temperature    """os_name=platform.system()ifos_name=='Darwin':fromsubprocessimportcheck_outputoutput=check_output(["sysctl","machdep.xcpm.cpu_thermal_level"])importrereturnre.findall(r'\d+',str(output))[0]else:returnpsutil.sensors_temperatures()["coretemp"][0]defline_protocol(temperature):"""Create a InfluxDB line protocol with structure:        iot_sensor,hostname=mine_sensor_12,type=temperature value=68    :param temperature: the sensor temperature    :return: Line protocol to write into InfluxDB    """importsocketreturn'iot_sensor,hostname={},type=temperature value={}'.format(socket.gethostname(),temperature)"""Read temperature every minute; distinct_until_changed - produce only if temperature change"""data=rx\    .interval(period=timedelta(seconds=60))\    .pipe(ops.map(lambdat:sensor_temperature()),ops.distinct_until_changed(),ops.map(lambdatemperature:line_protocol(temperature)))_db_client=InfluxDBClient(url="http://localhost:8086",token="my-token",org="my-org",debug=True)"""Create client that writes data into InfluxDB"""_write_api=_db_client.write_api(write_options=WriteOptions(batch_size=1))_write_api.write(bucket="my-bucket",record=data)"""Call after terminate a script"""atexit.register(on_exit,_db_client,_write_api)input()

Connect to InfluxDB Cloud

"""Connect to InfluxDB 2.0 - write data and query them"""fromdatetimeimportdatetimefrominfluxdb_clientimportPoint,InfluxDBClientfrominfluxdb_client.client.write_apiimportSYNCHRONOUS"""Configure credentials"""influx_cloud_url='https://us-west-2-1.aws.cloud2.influxdata.com'influx_cloud_token='...'bucket='...'org='...'client=InfluxDBClient(url=influx_cloud_url,token=influx_cloud_token)try:kind='temperature'host='host1'device='opt-123'"""    Write data by Point structure    """point=Point(kind).tag('host',host).tag('device',device).field('value',25.3).time(time=datetime.utcnow())print(f'Writing to InfluxDB cloud:{point.to_line_protocol()} ...')write_api=client.write_api(write_options=SYNCHRONOUS)write_api.write(bucket=bucket,org=org,record=point)print()print('success')print()print()"""    Query written data    """query=f'from(bucket: "{bucket}") |> range(start: -1d) |> filter(fn: (r) => r._measurement == "{kind}")'print(f'Querying from InfluxDB cloud: "{query}" ...')print()query_api=client.query_api()tables=query_api.query(query=query,org=org)fortableintables:forrowintable.records:print(f'{row.values["_time"]}: host={row.values["host"]},device={row.values["device"]} 'f'{row.values["_value"]} °C')print()print('success')exceptExceptionase:print(e)finally:client.close()

How to use Jupyter + Pandas + InfluxDB 2

Other examples

Advanced Usage

Gzip support

frominfluxdb_clientimportInfluxDBClient_db_client=InfluxDBClient(url="http://localhost:8086",token="my-token",org="my-org",enable_gzip=True)

Authenticate to the InfluxDB

Token

frominfluxdb_clientimportInfluxDBClientwithInfluxDBClient(url="http://localhost:8086",token="my-token")asclient

Note

Note that this is a preferred way how to authenticate to InfluxDB API.

Username & Password

frominfluxdb_clientimportInfluxDBClientwithInfluxDBClient(url="http://localhost:8086",username="my-user",password="my-password")asclient

Warning

Theusername/password auth is based on the HTTP "Basic" authentication.The authorization expires when thetime-to-live (TTL)(default 60 minutes) is reached and client producesunauthorized exception.

HTTP Basic

frominfluxdb_clientimportInfluxDBClientwithInfluxDBClient(url="http://localhost:8086",auth_basic=True,token="my-proxy-secret")asclient

Warning

Don't use this when directly talking to InfluxDB 2.

Proxy configuration

frominfluxdb_clientimportInfluxDBClientwithInfluxDBClient(url="http://localhost:8086",token="my-token",org="my-org",proxy="http://localhost:3128")asclient:

Note

If your proxy notify the client with permanent redirect (HTTP 301) todifferent host.The client removesAuthorization header, because otherwise the contents ofAuthorization is sent to third partieswhich is a security vulnerability.

You can change this behaviour by:

fromurllib3importRetryRetry.DEFAULT_REMOVE_HEADERS_ON_REDIRECT=frozenset()Retry.DEFAULT.remove_headers_on_redirect=Retry.DEFAULT_REMOVE_HEADERS_ON_REDIRECT

Delete data

frominfluxdb_clientimportInfluxDBClientclient=InfluxDBClient(url="http://localhost:8086",token="my-token")delete_api=client.delete_api()"""Delete Data"""start="1970-01-01T00:00:00Z"stop="2021-02-01T00:00:00Z"delete_api.delete(start,stop,'_measurement="my_measurement"',bucket='my-bucket',org='my-org')"""Close client"""client.close()

InfluxDB 1.8 API compatibility

Handling Errors

frominfluxdb_clientimportInfluxDBClientfrominfluxdb_client.client.exceptionsimportInfluxDBErrorfrominfluxdb_client.client.write_apiimportSYNCHRONOUSwithInfluxDBClient(url="http://localhost:8086",token="my-token",org="my-org")asclient:try:client.write_api(write_options=SYNCHRONOUS).write("my-bucket",record="mem,tag=a value=86")exceptInfluxDBErrorase:ife.response.status==401:raiseException(f"Insufficient write permissions to 'my-bucket'.")fromeraise

frominfluxdb_clientimportInfluxDBClientfrominfluxdb_client.client.exceptionsimportInfluxDBErrorclassBatchingCallback(object):defsuccess(self,conf: (str,str,str),data:str):print(f"Written batch:{conf}, data:{data}")deferror(self,conf: (str,str,str),data:str,exception:InfluxDBError):print(f"Cannot write batch:{conf}, data:{data} due:{exception}")defretry(self,conf: (str,str,str),data:str,exception:InfluxDBError):print(f"Retryable error occurs for batch:{conf}, data:{data} retry:{exception}")withInfluxDBClient(url="http://localhost:8086",token="my-token",org="my-org")asclient:callback=BatchingCallback()withclient.write_api(success_callback=callback.success,error_callback=callback.error,retry_callback=callback.retry)aswrite_api:pass

HTTP Retry Strategy

fromurllib3importRetryfrominfluxdb_clientimportInfluxDBClientretries=Retry(connect=5,read=2,redirect=5)client=InfluxDBClient(url="http://localhost:8086",token="my-token",org="my-org",retries=retries)

Nanosecond precision

frominfluxdb_clientimportPoint,InfluxDBClientfrominfluxdb_client.client.util.date_utils_pandasimportPandasDateTimeHelperfrominfluxdb_client.client.write_apiimportSYNCHRONOUS"""Set PandasDate helper which supports nanoseconds."""importinfluxdb_client.client.util.date_utilsasdate_utilsdate_utils.date_helper=PandasDateTimeHelper()"""Prepare client."""client=InfluxDBClient(url="http://localhost:8086",token="my-token",org="my-org")write_api=client.write_api(write_options=SYNCHRONOUS)query_api=client.query_api()"""Prepare data"""point=Point("h2o_feet") \    .field("water_level",10) \    .tag("location","pacific") \    .time('1996-02-25T21:20:00.001001231Z')print(f'Time serialized with nanosecond precision:{point.to_line_protocol()}')print()write_api.write(bucket="my-bucket",record=point)"""Query: using Stream"""query='''from(bucket:"my-bucket")        |> range(start: 0, stop: now())        |> filter(fn: (r) => r._measurement == "h2o_feet")'''records=query_api.query_stream(query)forrecordinrecords:print(f'Temperature in{record["location"]} is{record["_value"]} at time:{record["_time"]}')"""Close client"""client.close()

How to use Asyncio

Warning

TheInfluxDBClientAsync should be initialised insideasync coroutineotherwise there can be unexpected behaviour.For more info see:Why is creating a ClientSession outside of an event loop dangerous?.

Async APIs

Async Write API

Async Query API

Async Delete API

Management API

Proxy and redirects

frominfluxdb_client.client.influxdb_client_asyncimportInfluxDBClientAsyncasyncwithInfluxDBClientAsync(url="http://localhost:8086",token="my-token",org="my-org",proxy="http://localhost:3128")asclient:

Note

If your proxy notify the client with permanent redirect (HTTP 301) todifferent host.The client removesAuthorization header, because otherwise the contents ofAuthorization is sent to third partieswhich is a security vulnerability.

SQL Client Support

frominfluxdb_clientimportInfluxDBClientwithInfluxDBClient(url="http://localhost:8086",token="my-token",org="my-org",debug=False)asclient:sql_client=client.sql_client("my-bucket")reader=sql_client.query("select * from cpu limit 10")print(reader.read_all())

Warning

TheSQLClient only works with InfluxDB that has SQL support enabled.This does not apply to all InfluxDB versions.

Logging

importloggingimportsysfrominfluxdb_clientimportInfluxDBClientwithInfluxDBClient(url="http://localhost:8086",token="my-token",org="my-org")asclient:for_,loggerinclient.conf.loggers.items():logger.setLevel(logging.DEBUG)logger.addHandler(logging.StreamHandler(sys.stdout))

Debugging

client=InfluxDBClient(url="http://localhost:8086",token="my-token",debug=True)

Note

Both HTTP request headers and body will be logged to standard output.

Local tests

#start/restart InfluxDB2 onlocal machine using docker./scripts/influxdb-restart.sh#install requirementspip install -e . --userpip install -e .\[extra\] --userpip install -e .\[test\] --user#run unit& integration testspytest tests

Contributing

License