#2: Added support for synchronous and asynchronous writes

Author: bednar

README.md

@@ -2,10 +2,14 @@
 
 [![Build Status](https://travis-ci.org/bonitoo-io/influxdb-client-python.svg?branch=master)](https://travis-ci.org/bonitoo-io/influxdb-client-python)
 
-
 InfluxDB 2.0 python client library. TODO...
 
-## Requirements.
+- [Requirements](#requirements)
+- [Getting Started](#getting-started)
+- [How To Use](#how-to-use)
+    - [To write into InfluxDB 2.0](#writes)
+
+## Requirements
 
 Python 2.7 and 3.4+
 
@@ -45,12 +49,13 @@ Please follow the [installation procedure](#installation--usage) and then run th
 ```python
 from influxdb2.client.influxdb_client import InfluxDBClient
 from influxdb2.client.write.point import Point
+from influxdb2.client.write_api import SYNCHRONOUS
 
 bucket = "test_bucket"
 
 client = InfluxDBClient(url="http://localhost:9999/api/v2", token="my-token-123", org="my-org")
 
-write_api = client.write_api()
+write_api = client.write_api(write_options=SYNCHRONOUS)
 query_api = client.query_api()
 
 p = Point("my_measurement").tag("location", "Prague").field("temperature", 25.3)
@@ -72,7 +77,22 @@ val_count = 0
 for row in csv_result:
     for cell in row:
         val_count += 1
+```
 
+## How to use
 
+### Writes
 
-```
+The [WriteApiClient](https://github.com/bonitoo-io/influxdb-client-python/blob/master/influxdb2/client/write_api.py) that supports synchronous, asynchronous and batching writes into InfluxDB 2.0.
+
+The write client could be configured by `WriteOptions`:
+
+| Property | Description | Default Value |
+| --- | --- | --- |
+| [**write_type**](#write_type) | how the client writes data ; allowed values: `batching`, `asynchronous`, `synchronous`| `batching` |
+| **batch_size** | the number of data point to collect in batch | `1000` |
+
+##### write_type
+* `batching` - data are writes in batches defined by `batch_size`, `flush_interval`, ...
+* `asynchronous` - data are writes in asynchronous HTTP request
+* `synchronous` - data are writes in synchronous HTTP request

influxdb2/client/influxdb_client.py

@@ -6,7 +6,7 @@
 from influxdb2.client.organizations_api import OrganizationsApi
 from influxdb2.client.query_api import QueryApi
 from influxdb2.client.users_api import UsersApi
-from influxdb2.client.write_api import WriteApiClient
+from influxdb2.client.write_api import WriteApiClient, WriteOptions
 
 
 class InfluxDBClient(object):
@@ -38,10 +38,9 @@ def __init__(self,
         self.api_client = influxdb2.ApiClient(configuration=conf, header_name=auth_header_name,
                                               header_value=auth_header_value)
 
-    def write_api(self):
+    def write_api(self, write_options=WriteOptions()):
         service = influxdb2.service.write_service.WriteService(self.api_client)
-        return WriteApiClient(service=service)
-        # return
+        return WriteApiClient(service=service, write_options=write_options)
 
     def query_api(self):
         return QueryApi(self)

influxdb2/client/write_api.py

@@ -1,19 +1,22 @@
 # coding: utf-8
-
-
-from rx.scheduler import NewThreadScheduler
-from rx.subject import Subject
+from enum import Enum
 
 from influxdb2 import WritePrecision
 from influxdb2.client.abstract_client import AbstractClient
 from influxdb2.client.write.point import Point
 
 
+class WriteType(Enum):
+    batching = 1
+    asynchronous = 2
+    synchronous = 3
+
+
 class WriteOptions(object):
 
-    def __init__(self, batch_size=5000, flush_interval=1000, jitter_interval=0, retry_interval=1000,
-                 buffer_limit=10000,
-                 write_scheduler=NewThreadScheduler) -> None:
+    def __init__(self, write_type=WriteType.batching, batch_size=None, flush_interval=None, jitter_interval=None,
+                 retry_interval=None, buffer_limit=None, write_scheduler=None) -> None:
+        self.write_type = write_type
         self.batch_size = batch_size
         self.flush_interval = flush_interval
         self.jitter_interval = jitter_interval
@@ -22,13 +25,15 @@ def __init__(self, batch_size=5000, flush_interval=1000, jitter_interval=0, retr
         self.write_scheduler = write_scheduler
 
 
+SYNCHRONOUS = WriteOptions(write_type=WriteType.synchronous)
+ASYNCHRONOUS = WriteOptions(write_type=WriteType.asynchronous)
+
+
 class WriteApiClient(AbstractClient):
 
-    def __init__(self, service, write_options=None) -> None:
+    def __init__(self, service, write_options=WriteOptions()) -> None:
         self._write_service = service
-        self.write_options = write_options
-
-        _subject = Subject
+        self._write_options = write_options
 
     def write(self, bucket, org, record, write_precision=None):
 
@@ -52,8 +57,14 @@ def write(self, bucket, org, record, write_precision=None):
                     lines.append(item.to_line_protocol())
             final_string = '\n'.join(lines)
 
-        return self._write_service.post_write(org=org, bucket=bucket, body=final_string, precision=write_precision)
+        _async_req = True if self._write_options.write_type == WriteType.asynchronous else False
+
+        return self._write_service.post_write(org=org, bucket=bucket, body=final_string, precision=write_precision,
+                                              async_req=_async_req)
 
     def flush(self):
         # TODO
         pass
+
+    def __del__(self):
+        pass

influxdb2_test/base_test.py

@@ -26,9 +26,6 @@ def setUp(self) -> None:
         self.client = InfluxDBClient(conf.host, auth_token, debug=conf.debug, org="my-org")
         self.api_client = self.client.api_client
 
-        self.write_api = influxdb2.service.write_service.WriteService(self.api_client)
-        self.write_client = self.client.write_api()
-
         self.query_client = self.client.query_api()
         self.buckets_client = self.client.buckets_api()
         self.my_organization = self.find_my_org()

influxdb2_test/example.py

@@ -1,19 +1,16 @@
 import codecs
-import io
-import builtins
 from datetime import datetime
 
-import pandas
-
 from influxdb2 import WritePrecision
 from influxdb2.client.influxdb_client import InfluxDBClient
 from influxdb2.client.write.point import Point
+from influxdb2.client.write_api import SYNCHRONOUS
 
 bucket = "test_bucket"
 
 client = InfluxDBClient(url="http://localhost:9999/api/v2", token="my-token-123", org="my-org")
 
-write_api = client.write_api()
+write_api = client.write_api(write_options=SYNCHRONOUS)
 query_api = client.query_api()
 
 p = Point("my_measurement").tag("location", "Prague").field("temperature", 25.3).time(datetime.now(), WritePrecision.MS)

influxdb2_test/test_WriteApi.py

@@ -4,12 +4,22 @@
 
 import datetime
 import unittest
+from multiprocessing.pool import ApplyResult
 
 from influxdb2 import WritePrecision
+from influxdb2.client.write_api import SYNCHRONOUS, ASYNCHRONOUS
 from influxdb2_test.base_test import BaseTest
 
 
-class SimpleWriteTest(BaseTest):
+class SynchronousWriteTest(BaseTest):
+
+    def setUp(self) -> None:
+        super().setUp()
+        self.write_client = self.client.write_api(write_options=SYNCHRONOUS)
+
+    def tearDown(self) -> None:
+        self.write_client.__del__()
+        super().tearDown()
 
     def test_write_line_protocol(self):
         bucket = self.create_test_bucket()
@@ -32,11 +42,11 @@ def test_write_line_protocol(self):
 
     #####################################
 
-    def test_write_precission(self):
+    def test_write_precision(self):
         bucket = self.create_test_bucket()
 
-        self.client.write_api().write(org="my-org", bucket=bucket.name, record="air,location=Python humidity=99",
-                                      write_precision=WritePrecision.MS)
+        self.write_client.write(org="my-org", bucket=bucket.name, record="air,location=Python humidity=99",
+                                write_precision=WritePrecision.MS)
 
         result = self.query_client.query(
             "from(bucket:\"" + bucket.name + "\") |> range(start: 1970-01-01T00:00:00.000000001Z) |> last()", self.org)
@@ -45,17 +55,17 @@ def test_write_precission(self):
 
         self.delete_test_bucket(bucket)
 
-    def test_WriteRecordsList(self):
+    def test_write_records_list(self):
         bucket = self.create_test_bucket()
 
         _record1 = "h2o_feet,location=coyote_creek level\\ water_level=1.0 1"
         _record2 = "h2o_feet,location=coyote_creek level\\ water_level=2.0 2"
 
-        list = [_record1, _record2]
+        record_list = [_record1, _record2]
 
-        self.client.write_api().write(bucket.name, self.org, list)
+        self.write_client.write(bucket.name, self.org, record_list)
 
-        self.client.write_api().flush()
+        self.write_client.flush()
 
         query = 'from(bucket:"' + bucket.name + '") |> range(start: 1970-01-01T00:00:00.000000001Z)'
         print(query)
@@ -76,6 +86,36 @@ def test_WriteRecordsList(self):
         self.assertEqual(2, records[1].get_value())
         self.assertEqual("level water_level", records[1].get_field())
 
+    def test_write_result(self):
+
+        _bucket = self.create_test_bucket()
+
+        _record = "h2o_feet,location=coyote_creek level\\ water_level=1.0 1"
+        result = self.write_client.write(_bucket.name, self.org, _record)
+
+        # The success response is 204 - No Content
+        self.assertEqual(None, result)
+
+
+class AsynchronousWriteTest(BaseTest):
+
+    def setUp(self) -> None:
+        super().setUp()
+        self.write_client = self.client.write_api(write_options=ASYNCHRONOUS)
+
+    def tearDown(self) -> None:
+        self.write_client.__del__()
+        super().tearDown()
+
+    def test_write_result(self):
+        _bucket = self.create_test_bucket()
+
+        _record = "h2o_feet,location=coyote_creek level\\ water_level=1.0 1"
+        result = self.write_client.write(_bucket.name, self.org, _record)
+
+        self.assertEqual(ApplyResult, type(result))
+        self.assertEqual(None, result.get())
+
 
 if __name__ == '__main__':
     unittest.main()