destination/influxdb3: influxdb3sink time setter update (#607)

* update in preparation for updated influxdb3sink

* reorganize it and clean it up a bit

* add temporary changes to test PR

* tweak readme

* use main instead since it's merged

* change default influxdb url

Author: tim-quix

python/destinations/influxdb_3/README.md

@@ -2,6 +2,9 @@
 
 [This connector](https://github.com/quixio/quix-samples/tree/main/python/destinations/influxdb_3) demonstrates how to consume data from a Kafka topic in Quix and persist the data to an InfluxDB v3 database using the InfluxDB write API.
 
+To learn more about how it functions, [check out the underlying 
+Quix Streams `InfluxDB3Source`](https://quix.io/docs/quix-streams/connectors/sinks/influxdb3-sink.html).
+
 ## How to run
 
 Create a [Quix](https://portal.platform.quix.io/signup?xlink=github) account or log-in and visit the `Connectors` tab to use this connector.

python/destinations/influxdb_3/dockerfile

@@ -1,5 +1,12 @@
 FROM python:3.12.5-slim-bookworm
-			
+
+# TODO: remove this RUN block when done doing "@ git+" install in requirements.txt
+# This should be done BEFORE merging PR
+RUN apt-get update && \
+  apt-get install -y git && \
+  apt-get clean && \
+  rm -rf /var/lib/apt/lists/*
+
 # Set environment variables for non-interactive setup and unbuffered output
 ENV DEBIAN_FRONTEND=noninteractive \
     PYTHONUNBUFFERED=1 \

python/destinations/influxdb_3/library.json

@@ -26,7 +26,7 @@
       "Type": "EnvironmentVariable",
       "InputType": "FreeText",
       "Description": "Host address for the InfluxDB instance.",
-      "DefaultValue": "https://eu-central-1-1.aws.cloud2.influxdata.com",
+      "DefaultValue": "http://influxdb3:80",
       "Required": true
     },
     {

python/destinations/influxdb_3/main.py

@@ -1,6 +1,5 @@
 # import Utility modules
 import os
-import logging
 
 # import vendor-specific modules
 from quixstreams import Application
@@ -10,50 +9,35 @@
 from dotenv import load_dotenv
 load_dotenv()
 
-logging.basicConfig(level=logging.INFO)
-logger = logging.getLogger(__name__)
 
-# read the consumer group from config
-consumer_group_name = os.environ.get("CONSUMER_GROUP_NAME", "influxdb-data-writer")
-
-# read the timestamp column from config
-timestamp_column = os.environ.get("TIMESTAMP_COLUMN") if os.environ.get("TIMESTAMP_COLUMN") else None
+tag_keys = keys.split(",") if (keys := os.environ.get("INFLUXDB_TAG_KEYS")) else []
+field_keys = keys.split(",") if (keys := os.environ.get("INFLUXDB_FIELD_KEYS")) else []
+measurement_name = os.environ.get("INFLUXDB_MEASUREMENT_NAME", "measurement1")
+time_setter = col if (col := os.environ.get("TIMESTAMP_COLUMN")) else None
 
-buffer_size = int(os.environ.get("BUFFER_SIZE", "1000"))
+influxdb_v3_sink = InfluxDB3Sink(
+    token=os.environ["INFLUXDB_TOKEN"],
+    host=os.environ["INFLUXDB_HOST"],
+    organization_id=os.environ["INFLUXDB_ORG"],
+    tags_keys=tag_keys,
+    fields_keys=field_keys,
+    time_setter=time_setter,
+    database=os.environ["INFLUXDB_DATABASE"],
+    measurement=measurement_name,
+)
 
-buffer_delay = float(os.environ.get("BUFFER_DELAY", "1"))
 
-# Create a Quix platform-specific application instead
 app = Application(
-    consumer_group=consumer_group_name, 
+    consumer_group=os.environ.get("CONSUMER_GROUP_NAME", "influxdb-data-writer"),
     auto_offset_reset="earliest",
-    commit_every=buffer_size,
-    commit_interval=buffer_delay)
-    
+    commit_every=int(os.environ.get("BUFFER_SIZE", "1000")),
+    commit_interval=float(os.environ.get("BUFFER_DELAY", "1")),
+)
 input_topic = app.topic(os.environ["input"])
 
-# Read the environment variable and convert it to a dictionary
-tag_keys = os.environ.get("INFLUXDB_TAG_KEYS", "").split(",") if os.environ.get("INFLUXDB_TAG_KEYS") else []
-field_keys = os.environ.get("INFLUXDB_FIELD_KEYS", "").split(",")if os.environ.get("INFLUXDB_FIELD_KEYS") else []
-measurement_name = os.environ.get("INFLUXDB_MEASUREMENT_NAME", "measurement1")
-
-influxdb_v3_sink = InfluxDB3Sink(
-                            token=os.environ["INFLUXDB_TOKEN"],
-                            host=os.environ["INFLUXDB_HOST"],
-                            organization_id=os.environ["INFLUXDB_ORG"],
-                            tags_keys=tag_keys,
-                            fields_keys=field_keys,
-                            time_key=timestamp_column,
-                            database=os.environ["INFLUXDB_DATABASE"],
-                            measurement=measurement_name)
-
 sdf = app.dataframe(input_topic)
-
-#sdf.print()
 sdf.sink(influxdb_v3_sink)
 
+
 if __name__ == "__main__":
-    logger.info("Starting application")
     app.run()
-
-

python/destinations/influxdb_3/requirements.txt

@@ -1,2 +1,3 @@
-quixstreams[influxdb3]==3.8.1
+# TODO: remove "@ git+" version of install before merging PR
+quixstreams[influxdb3] @ git+https://github.com/quixio/quix-streams.git
 python-dotenv