questdb

import os import time from questdb.client import Sender from datetime import datetime, timezone # Configure QuestDB connection (ILP over HTTP) # Use os.environ.get for security and flexibility in deployment QDB_HOST = os.environ.get("QDB_HOST", "localhost") QDB_PORT = os.environ.get("QDB_PORT", "9000") QDB_AUTH = os.environ.get("QDB_AUTH", "") # Format: "username:token" conf = f'http::addr={QDB_HOST}:{QDB_PORT};' if QDB_AUTH: conf += f'auth={QDB_AUTH};' # Assumes auth syntax for QuestDB 7.0+ else: print("Warning: QDB_AUTH environment variable not set. Connecting without authentication.") try: with Sender.from_conf(conf) as sender: # Ingest a single row of data sender.row( "sensor_data", columns={ "location": "london", "temperature": 15.5, "humidity": 70, "event_time": datetime.now(timezone.utc) }, at=datetime.now(timezone.utc) # QuestDB timestamp column ) # Ingest another row, demonstrating nanosecond precision (client v4.0.0+) # Note: Requires QuestDB server 9.1.0+ for nanosecond TIMESTAMP_NS type nanos_timestamp = int(time.time_ns()) # Current time in nanoseconds sender.row( "sensor_data", columns={ "location": "paris", "temperature": 18.2, "humidity": 65, "event_time": datetime.now(timezone.utc) }, at=nanos_timestamp # This sends nanosecond precision ) sender.flush() # Ensure all buffered data is sent print("Data sent successfully to QuestDB!") except Exception as e: print(f"Failed to send data: {e}") print(f"Please ensure QuestDB server is running and accessible at http://{QDB_HOST}:{QDB_PORT}")