test: batch writing

tests/test_influxdb_client_3_integration.py

@@ -1,16 +1,16 @@
 import logging
 import os
-import pyarrow
-import pytest
 import random
 import string
 import time
 import unittest
 
+import pyarrow
+import pytest
 from urllib3.exceptions import MaxRetryError, TimeoutError as Url3TimeoutError
 
 from influxdb_client_3 import InfluxDBClient3, write_client_options, WriteOptions, \
-    WriteType, InfluxDB3ClientQueryError
+    WriteType, InfluxDB3ClientQueryError, Point
 from influxdb_client_3.exceptions import InfluxDBError
 from tests.util import asyncio_run, lp_to_py_object
 
@@ -40,14 +40,42 @@ def setUp(self):
         self.host = os.getenv('TESTING_INFLUXDB_URL')
         self.token = os.getenv('TESTING_INFLUXDB_TOKEN')
         self.database = os.getenv('TESTING_INFLUXDB_DATABASE')
-        self.client = InfluxDBClient3(host=self.host, database=self.database, token=self.token)
+        write_options=WriteOptions(batch_size=100)
+        # write_options=WriteOptions(write_type=WriteType.synchronous)
+        wco = write_client_options(write_options=write_options)
+        self.client = InfluxDBClient3(
+            host=self.host,
+            database=self.database,
+            token=self.token,
+            write_client_options=wco)
 
     def tearDown(self):
         self._caplog.clear()
         self._caplog.set_level(logging.ERROR)
         if self.client:
             self.client.close()
 
+    def test_write_batch_and_query(self):
+        # write_options=WriteOptions(batch_size=100)
+        # write_options=WriteOptions(write_type=WriteType.synchronous)
+        # wco = write_client_options(write_options=write_options)
+        # c = InfluxDBClient3(
+        #     host='http://localhost:8181',
+        #     database=self.database,
+        #     token='apiv3_XI_7D1lsUv0CsVoPiWQJtzG51LE-roQLF64pkKRxOi3cVdAFRfoBIJjptTjDordcOGa-rYe1Dow-iRsv9yU4ZA',
+        #     write_client_options=wco)
+
+        test_id = time.time_ns()
+        self.client.write(f"integration_test_python22,type=used value=123.0,test_id={test_id}i")
+
+        sql = 'SELECT * FROM integration_test_python22 where type=$type and test_id=$test_id'
+        df = self.client.query(sql, mode="pandas", query_parameters={'type': 'used', 'test_id': test_id})
+
+        self.assertIsNotNone(df)
+        self.assertEqual(1, len(df))
+        self.assertEqual(test_id, df['test_id'][0])
+        self.assertEqual(123.0, df['value'][0])
+
     def test_write_and_query(self):
         test_id = time.time_ns()
         self.client.write(f"integration_test_python,type=used value=123.0,test_id={test_id}i")