Merge pull request #39 from InfluxCommunity/38-write-error-when-org=-isnt-set-for-the-client

Pylint updated. fixed upload file flags

Author: Jayclifford345

Examples/file-import/write_file_parse_options.py

@@ -0,0 +1,46 @@
+import influxdb_client_3 as InfluxDBClient3
+import pandas as pd
+import numpy as np
+from influxdb_client_3 import write_client_options, WriteOptions, InfluxDBError, file_parser_options
+
+
+class BatchingCallback(object):
+
+    def success(self, conf, data: str):
+        print(f"Written batch: {conf}, data: {data}")
+
+    def error(self, conf, data: str, exception: InfluxDBError):
+        print(f"Cannot write batch: {conf}, data: {data} due: {exception}")
+
+    def retry(self, conf, data: str, exception: InfluxDBError):
+        print(f"Retryable error occurs for batch: {conf}, data: {data} retry: {exception}")
+
+callback = BatchingCallback()
+
+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)
+
+wco = write_client_options(success_callback=callback.success,
+                          error_callback=callback.error,
+                          retry_callback=callback.retry,
+                          WriteOptions=write_options 
+                        )
+
+with  InfluxDBClient3.InfluxDBClient3(
+    token="",
+    host="eu-central-1-1.aws.cloud2.influxdata.com",
+    org="6a841c0c08328fb1",
+    database="python", write_client_options=wco) as client:
+
+
+    fpo = file_parser_options(columns=["time", "machineID", "vibration"])
+
+    client.write_file(
+        file='./out.parquet',
+        timestamp_column='time', tag_columns=["provider", "machineID"], measurement_name='machine_data', file_parser_options=fpo)
+

influxdb_client_3/__init__.py

@@ -1,22 +1,42 @@
-import json
-import urllib.parse
+import urllib.parse, json
 import pyarrow as pa
 from influxdb_client import InfluxDBClient as _InfluxDBClient, WriteOptions, Point
 from influxdb_client.client.write_api import WriteApi as _WriteApi, SYNCHRONOUS, ASYNCHRONOUS, PointSettings
 from influxdb_client.domain.write_precision import WritePrecision
 from influxdb_client.client.exceptions import InfluxDBError
 from pyarrow.flight import FlightClient, Ticket, FlightCallOptions
-from influxdb_client_3.read_file import upload_file
+from influxdb_client_3.read_file import UploadFile
 
 
 def write_client_options(**kwargs):
+    """
+    Function for providing additional arguments for the WriteApi client.
+
+    :param kwargs: Additional arguments for the WriteApi client.
+    :return: dict with the arguments.
+    """
     return kwargs
 
 def default_client_options(**kwargs):
     return kwargs
 
 def flight_client_options(**kwargs):
-    return kwargs  # You can replace this with a specific data structure if needed
+    """
+    Function for providing additional arguments for the FlightClient.
+
+    :param kwargs: Additional arguments for the FlightClient.
+    :return: dict with the arguments.
+    """
+    return kwargs
+
+def file_parser_options(**kwargs):
+    """
+    Function for providing additional arguments for the file parser.
+
+    :param kwargs: Additional arguments for the file parser.
+    :return: dict with the arguments.
+    """
+    return kwargs  
 
 
 class InfluxDBClient3:
@@ -40,13 +60,13 @@ def __init__(
         :type database: str
         :param token: The authentication token for accessing the InfluxDB server.
         :type token: str
-        :param write_client_options: Options for the WriteAPI.
-        :type write_client_options: dict
-        :param flight_client_options: Options for the FlightClient.
-        :type flight_client_options: dict
+        :param write_client_options: Function for providing additional arguments for the WriteApi client.
+        :type write_client_options: callable
+        :param flight_client_options: Function for providing additional arguments for the FlightClient.
+        :type flight_client_options: callable
         :param kwargs: Additional arguments for the InfluxDB Client.
         """
-        self._org = org
+        self._org = org if org is not None else "default"
         self._database = database
         self._token = token
         self._write_client_options = write_client_options if write_client_options is not None else default_client_options(write_options=SYNCHRONOUS)
@@ -72,7 +92,9 @@ def write(self, record=None, database=None ,**kwargs):
         Write data to InfluxDB.
 
         :param record: The data point(s) to write.
-        :type record: Point or list of Point objects
+        :type record: object or list of objects
+        :param database: The database to write to. If not provided, uses the database provided during initialization.
+        :type database: str
         :param kwargs: Additional arguments to pass to the write API.
         """
         if database is None:
@@ -84,7 +106,7 @@ def write(self, record=None, database=None ,**kwargs):
             raise e
 
 
-    def write_file(self, file, measurement_name=None, tag_columns=None, timestamp_column='time', database=None, **kwargs):
+    def write_file(self, file, measurement_name=None, tag_columns=None, timestamp_column='time', database=None, file_parser_options=None ,**kwargs):
         """
         Write data from a file to InfluxDB.
 
@@ -96,13 +118,17 @@ def write_file(self, file, measurement_name=None, tag_columns=None, timestamp_co
         :type tag_columns: list
         :param timestamp_column: Timestamp column name. Defaults to 'time'.
         :type timestamp_column: str
+        :param database: The database to write to. If not provided, uses the database provided during initialization.
+        :type database: str
+        :param file_parser_options: Function for providing additional arguments for the file parser.
+        :type file_parser_options: callable
         :param kwargs: Additional arguments to pass to the write API.
         """
         if database is None:
             database = self._database
 
         try:
-            table = upload_file(file).load_file()
+            table = UploadFile(file, file_parser_options).load_file()
             df = table.to_pandas() if isinstance(table, pa.Table) else table
             self._process_dataframe(df, measurement_name, tag_columns or [], timestamp_column, database=database, **kwargs)
         except Exception as e:
@@ -138,10 +164,13 @@ def query(self, query, language="sql", mode="all", database=None, **kwargs ):
 
         :param query: The query string.
         :type query: str
-        :param language: The query language (default is "sql").
+        :param language: The query language; "sql" or "influxql" (default is "sql").
         :type language: str
         :param mode: The mode of fetching data (all, pandas, chunk, reader, schema).
         :type mode: str
+        :param database: The database to query from. If not provided, uses the database provided during initialization.
+        :type database: str
+        :param kwargs: Additional arguments for the query.
         :return: The queried data.
         """
         if database is None:
@@ -188,5 +217,6 @@ def __exit__(self, exc_type, exc_val, exc_tb):
     "WritePrecision",
     "WriteOptions",
     "write_client_options",
-    "flight_client_options"
+    "flight_client_options",
+    "file_parser_options"
 ]

influxdb_client_3/read_file.py

@@ -1,51 +1,99 @@
-from pyarrow import json as pa_json
+import os
 import pyarrow.csv as csv
 import pyarrow.feather as feather
 import pyarrow.parquet as parquet
-import os
+import pandas as pd
 
 # Check if the OS is not Windows
 if os.name != 'nt':
     import pyarrow.orc as orc
 
-import pandas as pd
 
-class upload_file:
-    def __init__(self, file, **kwargs):
+class UploadFile:
+    """
+    Class for uploading and reading different types of files.
+    """
+    def __init__(self, file, file_parser_options=None):
+        """
+        Initialize an UploadFile instance.
+
+        :param file: The file to upload.
+        :type file: str
+        :param kwargs: Additional arguments for file loading functions.
+        """
         self._file = file
-        self._kwargs = kwargs
+        self._kwargs = file_parser_options if file_parser_options is not None else {}
 
     def load_file(self):
-            if self._file.endswith(".feather"):
-                return self.load_feather(self._file, **self._kwargs)
-            elif self._file.endswith(".parquet"):
-                return self.load_parquet(self._file)
-            elif self._file.endswith(".csv"):
-                return self.load_csv(self._file)
-            elif self._file.endswith(".json"):
-                return self.load_json(self._file)
-            elif self._file.endswith(".orc"):
-
-                return self.load_orc(self._file)
-            else:
-                raise ValueError("Unsupported file type")
-
-    def load_feather(self, file):
+        """
+        Load a file based on its extension.
+
+        :return: The loaded file.
+        :raises ValueError: If the file type is not supported.
+        """
+        if self._file.endswith(".feather"):
+            return self.load_feather(self._file)
+        elif self._file.endswith(".parquet"):
+            return self.load_parquet(self._file)
+        elif self._file.endswith(".csv"):
+            return self.load_csv(self._file)
+        elif self._file.endswith(".json"):
+            return self.load_json(self._file)
+        elif self._file.endswith(".orc"):
+            return self.load_orc(self._file)
+        else:
+            raise ValueError("Unsupported file type")
+
+    def load_feather(self, file ):
+        """
+        Load a Feather file.
+
+        :param file: The Feather file to load.
+        :type file: str
+        :return: The loaded Feather file.
+        """
         return feather.read_table(file, **self._kwargs)
-        
+
     def load_parquet(self, file):
+        """
+        Load a Parquet file.
+
+        :param file: The Parquet file to load.
+        :type file: str
+        :return: The loaded Parquet file.
+        """
         return parquet.read_table(file, **self._kwargs)
-        
+
     def load_csv(self, file):
+        """
+        Load a CSV file.
+
+        :param file: The CSV file to load.
+        :type file: str
+        :return: The loaded CSV file.
+        """
         return csv.read_csv(file, **self._kwargs)
-    
+
     def load_orc(self, file):
+        """
+        Load an ORC file.
+
+        :param file: The ORC file to load.
+        :type file: str
+        :return: The loaded ORC file.
+        :raises ValueError: If the OS is Windows.
+        """
         if os.name == 'nt':
             raise ValueError("Unsupported file type for this OS")
         else:
             return orc.read_table(file, **self._kwargs)
-    
-    #TODO: Use pyarrow.json.read_json() instead of pandas.read_json()
+
     def load_json(self, file):
-        return pd.read_json(file, **self._kwargs)
-    
+        """
+        Load a JSON file.
+
+        :param file: The JSON file to load.
+        :type file: str
+        :return: The loaded JSON file.
+        """
+        return pd.read_json(file, **self._kwargs)