Initialize client library from config file and environmental properties (#52)

Author: rolincova

CHANGELOG.md

@@ -1,7 +1,8 @@
 ## 1.3.0 [unreleased]
 
 ### Features
-1. [#49](https://github.com/influxdata/influxdb-client-python/issues/50): Implemented default tags
+1. [#50](https://github.com/influxdata/influxdb-client-python/issues/50): Implemented default tags
+2. [#52](https://github.com/influxdata/influxdb-client-python/issues/52): Initialize client library from config file and environmental properties
 
 ### API
 1. [#47](https://github.com/influxdata/influxdb-client-python/pull/47): Updated swagger to latest version

README.rst

@@ -136,7 +136,49 @@ Please follow the `Installation`_ and then run the following:
        for cell in row:
            val_count += 1
 
+
 .. marker-query-end
+
+Client configuration
+--------------------
+
+Via File
+^^^^^^^^
+A client can be configured via ``*.ini`` file in segment ``influx2``.
+
+The following options are supported:
+
+- ``url`` - the url to connect to InfluxDB
+- ``org`` - default destination organization for writes and queries
+- ``token`` - the token to use for the authorization
+- ``timeout`` - socket timeout in ms (default value is 10000)
+
+.. code-block:: python
+
+    self.client = InfluxDBClient.from_config_file("config.ini")
+
+.. code-block::
+
+    [influx2]
+    url=http://localhost:9999
+    org=my-org
+    token=my-token
+    timeout=6000
+
+Via Environment Properties
+^^^^^^^^^^^^^^^^^^^^^^^^^^
+A client can be configured via environment properties.
+
+Supported properties are:
+
+- ``INFLUXDB_V2_URL`` - the url to connect to InfluxDB
+- ``INFLUXDB_V2_ORG`` - default destination organization for writes and queries
+- ``INFLUXDB_V2_TOKEN`` - the token to use for the authorization
+- ``INFLUXDB_V2_TIMEOUT`` - socket timeout in ms (default value is 10000)
+
+.. code-block:: python
+    self.client = InfluxDBClient.from_env_properties()
+
 .. marker-index-end
 
 
@@ -264,6 +306,9 @@ The expressions:
 - ``California Miner`` - static value
 - ``${env.hostname}`` - environment property
 
+Via API
+_______
+
 .. code-block:: python
 
     point_settings = PointSettings()
@@ -278,6 +323,42 @@ The expressions:
     self.write_client = self.client.write_api(write_options=SYNCHRONOUS,
                                                   point_settings=PointSettings(**{"id": "132-987-655",
                                                                                   "customer": "California Miner"}))
+
+Via Configuration file
+______________________
+
+In a ini configuration file you are able to specify default tags by ``tags`` segment.
+
+.. code-block:: python
+
+    self.client = InfluxDBClient.from_config_file("config.ini")
+
+.. code-block::
+
+    [influx2]
+    url=http://localhost:9999
+    org=my-org
+    token=my-token
+    timeout=6000
+
+    [tags]
+    id = 132-987-655
+    customer = California Miner
+    data_center = ${env.data_center}
+
+Via Environment Properties
+__________________________
+You are able to specify default tags by environment properties with prefix ``INFLUXDB_V2_TAG_``.
+
+Examples:
+
+- ``INFLUXDB_V2_TAG_ID``
+- ``INFLUXDB_V2_TAG_HOSTNAME``
+
+.. code-block:: python
+
+    self.client = InfluxDBClient.from_env_properties()
+
 .. marker-default-tags-end
 
 Asynchronous client

influxdb_client/client/influxdb_client.py

@@ -1,5 +1,8 @@
 from __future__ import absolute_import
 
+import configparser
+import os
+
 from influxdb_client import Configuration, ApiClient, HealthCheck, HealthService, Ready, ReadyService
 from influxdb_client.client.authorizations_api import AuthorizationsApi
 from influxdb_client.client.bucket_api import BucketsApi
@@ -14,7 +17,8 @@
 
 class InfluxDBClient(object):
 
-    def __init__(self, url, token, debug=None, timeout=10000, enable_gzip=False, org: str = None) -> None:
+    def __init__(self, url, token, debug=None, timeout=10000, enable_gzip=False, org: str = None,
+                 default_tags: dict = None) -> None:
         """
         :class:`influxdb_client.InfluxDBClient` is client for HTTP API defined
         in https://github.com/influxdata/influxdb/blob/master/http/swagger.yml.
@@ -33,6 +37,8 @@ def __init__(self, url, token, debug=None, timeout=10000, enable_gzip=False, org
         self.timeout = timeout
         self.org = org
 
+        self.default_tags = default_tags
+
         conf = _Configuration()
         conf.host = self.url
         conf.enable_gzip = enable_gzip
@@ -45,6 +51,50 @@ def __init__(self, url, token, debug=None, timeout=10000, enable_gzip=False, org
         self.api_client = ApiClient(configuration=conf, header_name=auth_header_name,
                                     header_value=auth_header_value)
 
+    @classmethod
+    def from_config_file(cls, config_file: str = "config.ini", debug=None, enable_gzip=False):
+        config = configparser.ConfigParser()
+        config.read(config_file)
+
+        url = config['influx2']['url']
+        token = config['influx2']['token']
+
+        timeout = None
+
+        if config.has_option('influx2', 'timeout'):
+            timeout = config['influx2']['timeout']
+
+        org = None
+
+        if config.has_option('influx2', 'org'):
+            org = config['influx2']['org']
+
+        default_tags = None
+
+        if config.has_section('tags'):
+            default_tags = dict(config.items('tags'))
+
+        if timeout:
+            return cls(url, token, debug=debug, timeout=int(timeout), org=org, default_tags=default_tags,
+                       enable_gzip=enable_gzip)
+
+        return cls(url, token, debug=debug, org=org, default_tags=default_tags, enable_gzip=enable_gzip)
+
+    @classmethod
+    def from_env_properties(cls, debug=None, enable_gzip=False):
+        url = os.getenv('INFLUXDB_V2_URL', "http://localhost:9999")
+        token = os.getenv('INFLUXDB_V2_TOKEN', "my-token")
+        timeout = os.getenv('INFLUXDB_V2_TIMEOUT', "10000")
+        org = os.getenv('INFLUXDB_V2_ORG', "my-org")
+
+        default_tags = dict()
+
+        for key, value in os.environ.items():
+            if key.startswith("INFLUXDB_V2_TAG_"):
+                default_tags[key[16:].lower()] = value
+
+        return cls(url, token, debug=debug, timeout=int(timeout), org=org, default_tags=default_tags, enable_gzip=enable_gzip)
+
     def write_api(self, write_options=WriteOptions(), point_settings=PointSettings()) -> WriteApi:
         """
         Creates a Write API instance

influxdb_client/client/write_api.py

@@ -6,6 +6,7 @@
 from time import sleep
 from typing import Union, List
 
+import configparser
 import os
 
 import rx
@@ -138,6 +139,11 @@ def __init__(self, influxdb_client, write_options: WriteOptions = WriteOptions()
         self._write_service = WriteService(influxdb_client.api_client)
         self._write_options = write_options
         self._point_settings = point_settings
+
+        if influxdb_client.default_tags:
+            for key, value in influxdb_client.default_tags.items():
+                self._point_settings.add_default_tag(key, value)
+
         if self._write_options.write_type is WriteType.batching:
             # Define Subject that listen incoming data and produces writes into InfluxDB
             self._subject = Subject()

tests/config.ini

@@ -0,0 +1,10 @@
+[influx2]
+url=http://localhost:9999
+org=my-org
+token=my-token
+timeout=6000
+
+[tags]
+id = 132-987-655
+customer = California Miner
+data_center = ${env.data_center}

tests/test_WriteApi.py

@@ -8,7 +8,7 @@
 import time
 from multiprocessing.pool import ApplyResult
 
-from influxdb_client import Point, WritePrecision
+from influxdb_client import Point, WritePrecision, InfluxDBClient
 from influxdb_client.client.write_api import SYNCHRONOUS, ASYNCHRONOUS, PointSettings
 from influxdb_client.rest import ApiException
 from tests.base_test import BaseTest
@@ -444,5 +444,93 @@ def test_point_settings_with_add(self):
         self.assertEqual("LA", default_tags.get("data_center"))
 
 
+class DefaultTagsConfiguration(BaseTest):
+
+    def setUp(self) -> None:
+        super().setUp()
+
+        os.environ['data_center'] = "LA"
+
+        self.id_tag = "132-987-655"
+        self.customer_tag = "California Miner"
+        self.data_center_key = "data_center"
+
+        os.environ['INFLUXDB_V2_URL'] = "http://localhost:9999"
+        os.environ['INFLUXDB_V2_TOKEN'] = "my-token"
+        os.environ['INFLUXDB_V2_TIMEOUT'] = "6000"
+        os.environ['INFLUXDB_V2_ORG'] = "my-org"
+
+        os.environ['INFLUXDB_V2_TAG_ID'] = self.id_tag
+        os.environ['INFLUXDB_V2_TAG_CUSTOMER'] = self.customer_tag
+        os.environ['INFLUXDB_V2_TAG_DATA_CENTER'] = "${env.data_center}"
+
+    def tearDown(self) -> None:
+        self.write_client.__del__()
+        super().tearDown()
+
+    def test_connection_option_from_conf_file(self):
+        self.client.close()
+        self.client = InfluxDBClient.from_config_file("config.ini", self.debug)
+
+        self._check_connection_settings()
+
+    def test_connection_option_from_env(self):
+        self.client.close()
+        self.client = InfluxDBClient.from_env_properties(self.debug)
+
+        self._check_connection_settings()
+
+    def _check_connection_settings(self):
+        self.write_client = self.client.write_api(write_options=SYNCHRONOUS)
+
+        self.assertEqual("http://localhost:9999", self.client.url)
+        self.assertEqual("my-org", self.client.org)
+        self.assertEqual("my-token", self.client.token)
+        self.assertEqual(6000, self.client.timeout)
+
+    def test_default_tags_from_conf_file(self):
+        self.client.close()
+        self.client = InfluxDBClient.from_config_file("config.ini", self.debug)
+
+        self._write_point()
+
+    def test_default_tags_from_env(self):
+        self.client.close()
+        self.client = InfluxDBClient.from_env_properties(self.debug)
+
+        self._write_point()
+
+    def _write_point(self):
+        self.write_client = self.client.write_api(write_options=SYNCHRONOUS)
+
+        bucket = self.create_test_bucket()
+
+        measurement = "h2o_feet"
+        field_name = "water_level"
+        val = "1.0"
+        tag = "location"
+        tag_value = "creek level"
+
+        p = Point(measurement)
+        p.field(field_name, val)
+        p.tag(tag, tag_value)
+
+        record_list = [p]
+        print(record_list)
+
+        self.write_client.write(bucket.name, self.org, record_list)
+
+        query = 'from(bucket:"' + bucket.name + '") |> range(start: 1970-01-01T00:00:00.000000001Z)'
+        flux_result = self.client.query_api().query(query)
+        self.assertEqual(1, len(flux_result))
+        rec = flux_result[0].records[0]
+
+        self.assertEqual(self.id_tag, rec["id"])
+        self.assertEqual(self.customer_tag, rec["customer"])
+        self.assertEqual("LA", rec[self.data_center_key])
+
+        self.delete_test_bucket(bucket)
+
+
 if __name__ == '__main__':
     unittest.main()