Merge pull request #18 from InfluxCommunity/17-add-ability-to-expose-batching-options

exposed write options

Author: Jayclifford345

Examples/cloud_dedicated_example.py renamed to Examples/cloud_dedicated_query.py

@@ -5,7 +5,7 @@
 client = InfluxDBClient3.InfluxDBClient3(token="",
                          host="b0c7cce5-8dbc-428e-98c6-7f996fb96467.a.influxdb.io",
                          org="6a841c0c08328fb1",
-                         database="flight2", write_options="SYNCHRONOUS")
+                         database="flight2")
 
 
 table = client.query(query="SELECT * FROM flight WHERE time > now() - 4h", language="influxql")

Examples/cloud_dedicated_write.py

@@ -0,0 +1,39 @@
+
+import influxdb_client_3 as InfluxDBClient3
+from influxdb_client_3 import write_options
+import pandas as pd
+import numpy as np
+
+
+
+client = InfluxDBClient3.InfluxDBClient3(token="",
+                         host="b0c7cce5-8dbc-428e-98c6-7f996fb96467.a.influxdb.io",
+                         org="6a841c0c08328fb1",
+                         database="flight2", write_options=write_options(batch_size=500,
+                                                      flush_interval=10_000,
+                                                      jitter_interval=2_000,
+                                                      retry_interval=5_000,
+                                                      max_retries=5,
+                                                      max_retry_delay=30_000,
+                                                      max_close_wait=300_000,
+                                                      exponential_base=2, write_type='batching'))
+
+
+
+
+
+# Create a dataframe
+df = pd.DataFrame({'a': [1, 2, 3], 'b': [4, 5, 6]})
+
+
+# Create a range of datetime values
+dates = pd.date_range(start='2023-05-01', end='2023-05-29', freq='5min')
+
+# Create a DataFrame with random data and datetime index
+df = pd.DataFrame(np.random.randn(len(dates), 3), index=dates, columns=['Column 1', 'Column 2', 'Column 3'])
+df['tagkey'] = 'Hello World'
+
+print(df)
+
+# Write the DataFrame to InfluxDB
+client.write(df, data_frame_measurement_name='table', data_frame_tag_columns=['tagkey']) 

Examples/csv_write.py

@@ -2,10 +2,10 @@
 import pandas as pd
 import numpy as np
 
-client = InfluxDBClient3.InfluxDBClient3(token="pjvlpe-olsMjvdmw8pejTIqkknI2KUGIUG-mIrrT824Y7H_tMkSwPqXn6IoEKnwqoGU8WznsJJWk4N5uI8MBsw==",
+client = InfluxDBClient3.InfluxDBClient3(token="",
                          host="eu-central-1-1.aws.cloud2.influxdata.com",
                          org="6a841c0c08328fb1",
-                         database="test", write_options="SYNCHRONOUS")
+                         database="test")
 
 
 

influxdb_client_3/__init__.py

@@ -3,13 +3,19 @@
 from pyarrow import csv
 from pyarrow.flight import FlightClient, Ticket, FlightCallOptions
 from influxdb_client import InfluxDBClient as _InfluxDBClient
+from influxdb_client import WriteOptions as _WriteOptions
 from influxdb_client.client.write_api import WriteApi as _WriteApi
 from influxdb_client.client.write_api import SYNCHRONOUS, ASYNCHRONOUS
 from influxdb_client import Point
 import json
 
+def write_options(**kwargs):
+   return  _WriteOptions(**kwargs)
+
+
+
 class InfluxDBClient3:
-    def __init__(self, host=None, org=None, database=None, token=None, write_options="SYNCHRONOUS", **kwargs):
+    def __init__(self, host=None, org=None, database=None, token=None, write_options=None, **kwargs):
         """
         This class provides an interface for interacting with an InfluxDB server, simplifying common operations such as writing, querying.
         * host (str, optional): The hostname or IP address of the InfluxDB server. Defaults to None.
@@ -21,13 +27,10 @@ def __init__(self, host=None, org=None, database=None, token=None, write_options
         """
         self._org = org
         self._database = database
+        self.write_options = write_options
 
-        if write_options == "SYNCHRONOUS":
+        if self.write_options == None:
             self.write_options = SYNCHRONOUS
-        elif write_options == "ASYNCHRONOUS":
-            self.write_options = ASYNCHRONOUS
-        else:
-            raise ValueError("write_options must be SYNCHRONOUS or ASYNCHRONOUS")
 
 
         self._client = _InfluxDBClient(url=f"https://{host}", token=token, org=self._org, **kwargs )